Clear events queue on rollback (single and bulk changes)

netbox/extras/tests/test_event_rules.py

@@ -19,8 +19,8 @@ from rest_framework import status
 from core.choices import ManagedFileRootPathChoices
 from core.events import *
 from core.models import Job, ObjectType
-from dcim.choices import SiteStatusChoices
-from dcim.models import DeviceType, Interface, Manufacturer, Site
+from dcim.choices import DeviceStatusChoices, InterfaceTypeChoices, SiteStatusChoices
+from dcim.models import Device, DeviceRole, DeviceType, Interface, Manufacturer, Site
 from extras.choices import EventRuleActionChoices
 from extras.events import enqueue_event, flush_events, process_event_rules, serialize_for_event
 from extras.models import EventRule, Script, ScriptModule, Tag, Webhook
@@ -36,7 +36,8 @@ from netbox.event_rules import (
 )
 from netbox.registry import registry
 from netbox.tests.dummy_plugin.event_rules import DummyRaisingAction
-from utilities.testing import APITestCase, create_test_device
+from users.models import ObjectPermission
+from utilities.testing import APITestCase, create_test_device, disable_warnings
 from utilities.testing.mixins import RQQueueTestMixin
 
 
@@ -218,6 +219,32 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
         self.assertEqual(job.kwargs['snapshots']['postchange']['name'], 'Site 1')
         self.assertEqual(job.kwargs['snapshots']['postchange']['tags'], ['Bar', 'Foo'])
 
+    def test_single_create_rollback_discards_events(self):
+        """
+        Check that creating an object which is then rolled back by the object-level permission check
+        in perform_create() queues no background task.
+        """
+        # Permit the creation of active sites only. The new object is saved (queueing its event)
+        # before _validate_objects() rejects it and the transaction is rolled back.
+        obj_perm = ObjectPermission(
+            name='Test permission',
+            actions=['add'],
+            constraints={'status': SiteStatusChoices.STATUS_ACTIVE},
+        )
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(Site))
+
+        data = {'name': 'Site 1', 'slug': 'site-1', 'status': SiteStatusChoices.STATUS_PLANNED}
+        url = reverse('dcim-api:site-list')
+        with disable_warnings('django.request'):
+            response = self.client.post(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
+        self.assertEqual(Site.objects.count(), 0)
+
+        # No task may be queued for a creation that was rolled back
+        self.assertEqual(self.queue.count, 0)
+
     def test_bulk_create_process_eventrule(self):
         """
         Check that bulk creating multiple objects with an applicable EventRule queues a background task for each
@@ -269,6 +296,40 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
             self.assertEqual(job.kwargs['snapshots']['postchange']['name'], response.data[i]['name'])
             self.assertEqual(job.kwargs['snapshots']['postchange']['tags'], ['Bar', 'Foo'])
 
+    def test_bulk_create_rollback_discards_events(self):
+        """
+        Check that a sequential bulk create which is rolled back queues no background tasks for the
+        objects that were provisionally created before the failure.
+        """
+        manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1')
+        device_type = DeviceType.objects.create(manufacturer=manufacturer, model='Device Type 1', slug='device-type-1')
+        role = DeviceRole.objects.create(name='Device Role 1', slug='device-role-1')
+        site = Site.objects.create(name='Site 1', slug='site-1')
+
+        # DeviceViewSet uses SequentialBulkCreatesMixin, so each valid object is provisionally
+        # created (and its event queued) before a later object fails validation.
+        event_rule = EventRule.objects.get(name='Event Rule 1')
+        event_rule.object_types.set([ObjectType.objects.get_for_model(Device)])
+
+        data = [
+            {
+                'name': 'Device 1',
+                'device_type': device_type.pk,
+                'role': role.pk,
+                'site': site.pk,
+                'status': DeviceStatusChoices.STATUS_ACTIVE,
+            },
+            {},  # Missing all required fields
+        ]
+        url = reverse('dcim-api:device-list')
+        self.add_permissions('dcim.add_device', 'dcim.view_site', 'dcim.view_devicetype', 'dcim.view_devicerole')
+        response = self.client.post(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(Device.objects.count(), 0)
+
+        # No task may be queued for a creation that was rolled back
+        self.assertEqual(self.queue.count, 0)
+
     def test_single_update_process_eventrule(self):
         """
         Check that updating an object with an applicable EventRule queues a background task for the rule's action.
@@ -303,6 +364,37 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
         self.assertEqual(job.kwargs['snapshots']['postchange']['name'], 'Site X')
         self.assertEqual(job.kwargs['snapshots']['postchange']['tags'], ['Baz'])
 
+    def test_single_update_rollback_discards_events(self):
+        """
+        Check that updating an object which is then rolled back by the object-level permission check
+        in perform_update() queues no background task.
+        """
+        site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_ACTIVE)
+
+        # Permit the modification of active sites only. Setting the status to "planned" takes the
+        # object outside the permission's scope, so it is saved (queueing its event) and then
+        # rejected by _validate_objects(), rolling the transaction back.
+        obj_perm = ObjectPermission(
+            name='Test permission',
+            actions=['change'],
+            constraints={'status': SiteStatusChoices.STATUS_ACTIVE},
+        )
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(Site))
+
+        url = reverse('dcim-api:site-detail', kwargs={'pk': site.pk})
+        with disable_warnings('django.request'):
+            response = self.client.patch(
+                url, {'status': SiteStatusChoices.STATUS_PLANNED}, format='json', **self.header
+            )
+        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
+        site.refresh_from_db()
+        self.assertEqual(site.status, SiteStatusChoices.STATUS_ACTIVE)
+
+        # No task may be queued for an update that was rolled back
+        self.assertEqual(self.queue.count, 0)
+
     def test_bulk_update_process_eventrule(self):
         """
         Check that bulk updating multiple objects with an applicable EventRule queues a background task for each
@@ -360,6 +452,38 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
             self.assertEqual(job.kwargs['snapshots']['postchange']['name'], response.data[i]['name'])
             self.assertEqual(job.kwargs['snapshots']['postchange']['tags'], ['Baz'])
 
+    def test_bulk_update_rollback_discards_events(self):
+        """
+        Check that a bulk update which is rolled back because one object failed validation queues no
+        background tasks for the objects that were provisionally updated.
+        """
+        sites = (
+            Site(name='Site 1', slug='site-1'),
+            Site(name='Site 2', slug='site-2'),
+            Site(name='Site 3', slug='site-3'),
+        )
+        Site.objects.bulk_create(sites)
+
+        # The first two objects are valid and will be provisionally updated; the third fails
+        # validation, rolling the entire batch back.
+        data = [
+            {'id': sites[0].pk, 'name': 'Site X'},
+            {'id': sites[1].pk, 'name': 'Site Y'},
+            {'id': sites[2].pk, 'status': 'not-a-valid-status'},
+        ]
+        url = reverse('dcim-api:site-list')
+        self.add_permissions('dcim.change_site')
+        response = self.client.patch(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+
+        # No object may have been modified
+        for site in sites:
+            site.refresh_from_db()
+        self.assertListEqual([site.name for site in sites], ['Site 1', 'Site 2', 'Site 3'])
+
+        # No task may be queued for an update that was rolled back
+        self.assertEqual(self.queue.count, 0)
+
     def test_single_delete_process_eventrule(self):
         """
         Check that deleting an object with an applicable EventRule queues a background task for the rule's action.
@@ -384,6 +508,35 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
         self.assertEqual(job.kwargs['snapshots']['prechange']['name'], 'Site 1')
         self.assertEqual(job.kwargs['snapshots']['prechange']['tags'], ['Bar', 'Foo'])
 
+    def test_single_delete_rollback_discards_events(self):
+        """
+        Check that deleting an object whose cascading deletion is aborted queues no background task
+        for the dependent objects that were already processed.
+        """
+        device = create_test_device('Device 1')
+        Interface.objects.create(
+            device=device, name='Interface 1', type=InterfaceTypeChoices.TYPE_1GE_FIXED, description='Has one'
+        )
+        Interface.objects.create(device=device, name='Interface 2', type=InterfaceTypeChoices.TYPE_1GE_FIXED)
+
+        event_rule = EventRule.objects.get(name='Event Rule 3')
+        event_rule.object_types.set([ObjectType.objects.get_for_model(Interface)])
+
+        url = reverse('dcim-api:device-detail', kwargs={'pk': device.pk})
+        self.add_permissions('dcim.delete_device')
+
+        # Deleting the Device cascades to both Interfaces. The first satisfies the protection rule
+        # and so is processed (queueing its event); the second does not, aborting the request.
+        protection_rules = {'dcim.interface': [{'description': {'required': True}}]}
+        with override_settings(PROTECTION_RULES=protection_rules):
+            response = self.client.delete(url, **self.header)
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertTrue(Device.objects.filter(pk=device.pk).exists())
+        self.assertEqual(Interface.objects.filter(device=device).count(), 2)
+
+        # No task may be queued for a deletion that was rolled back
+        self.assertEqual(self.queue.count, 0)
+
     def test_bulk_delete_process_eventrule(self):
         """
         Check that bulk deleting multiple objects with an applicable EventRule queues a background task for each
@@ -418,6 +571,58 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
             self.assertEqual(job.kwargs['snapshots']['prechange']['name'], sites[i].name)
             self.assertEqual(job.kwargs['snapshots']['prechange']['tags'], ['Bar', 'Foo'])
 
+    def test_bulk_delete_rollback_discards_events(self):
+        """
+        Check that a bulk delete which is rolled back because one object is protected queues no
+        background tasks for the objects that were provisionally deleted.
+        """
+        sites = (
+            Site(name='Site 1', slug='site-1'),
+            Site(name='Site 2', slug='site-2'),
+            Site(name='Site 3', slug='site-3'),
+        )
+        Site.objects.bulk_create(sites)
+
+        # A Device references the third Site, whose deletion will therefore raise a ProtectedError
+        # and roll the entire batch back.
+        create_test_device('Device 1', site=sites[2])
+
+        data = [{'id': site.pk} for site in sites]
+        url = reverse('dcim-api:site-list')
+        self.add_permissions('dcim.delete_site')
+        response = self.client.delete(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_409_CONFLICT)
+        self.assertEqual(Site.objects.count(), 3)
+
+        # No task may be queued for a deletion that was rolled back
+        self.assertEqual(self.queue.count, 0)
+
+    def test_bulk_delete_abort_discards_events(self):
+        """
+        Check that a bulk delete aborted by an exception (rather than by a per-object error) also
+        queues no background tasks. A protection rule raises AbortRequest from a signal receiver,
+        which propagates out of the per-object loop.
+        """
+        sites = (
+            Site(name='Site 1', slug='site-1', description='Has a description'),
+            Site(name='Site 2', slug='site-2'),
+        )
+        Site.objects.bulk_create(sites)
+
+        data = [{'id': site.pk} for site in sites]
+        url = reverse('dcim-api:site-list')
+        self.add_permissions('dcim.delete_site')
+
+        # Site 2 has no description, so its deletion is blocked once Site 1 has already been deleted
+        protection_rules = {'dcim.site': [{'description': {'required': True}}]}
+        with override_settings(PROTECTION_RULES=protection_rules):
+            response = self.client.delete(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(Site.objects.count(), 2)
+
+        # No task may be queued for a deletion that was rolled back
+        self.assertEqual(self.queue.count, 0)
+
     @skipIf('netbox.tests.dummy_plugin' not in settings.PLUGINS, 'dummy_plugin not in settings.PLUGINS')
     def test_send_webhook(self):
         request_id = uuid.uuid4()

netbox/netbox/api/viewsets/__init__.py

@@ -284,8 +284,9 @@ class NetBoxModelViewSet(
         logger.info(f"Creating new {model._meta.verbose_name}")
 
         # Enforce object-level permissions on save()
+        using = router.db_for_write(model)
         try:
-            with transaction.atomic(using=router.db_for_write(model)):
+            with transaction.atomic(using=using), mixins.discard_events_on_rollback(self, using=using):
                 instance = serializer.save()
                 self._validate_objects(instance)
         except ObjectDoesNotExist:
@@ -323,8 +324,9 @@ class NetBoxModelViewSet(
         logger.info(f"Updating {model._meta.verbose_name} {serializer.instance} (PK: {serializer.instance.pk})")
 
         # Enforce object-level permissions on save()
+        using = router.db_for_write(model)
         try:
-            with transaction.atomic(using=router.db_for_write(model)):
+            with transaction.atomic(using=using), mixins.discard_events_on_rollback(self, using=using):
                 # Re-check the If-Match ETag under a row-level lock to close the TOCTOU window
                 # between the initial check in update() and the actual write.
                 if self._get_if_match(self.request):
@@ -357,8 +359,9 @@ class NetBoxModelViewSet(
         logger = logging.getLogger(f'netbox.api.views.{self.__class__.__name__}')
         logger.info(f"Deleting {model._meta.verbose_name} {instance} (PK: {instance.pk})")
 
+        using = router.db_for_write(model)
         try:
-            with transaction.atomic(using=router.db_for_write(model)):
+            with transaction.atomic(using=using), mixins.discard_events_on_rollback(self, using=using):
                 # Re-check the If-Match ETag under a row-level lock to close the TOCTOU window
                 # between the initial check in destroy() and the actual delete.
                 if self._get_if_match(self.request):

netbox/netbox/api/viewsets/mixins.py

@@ -1,3 +1,5 @@
+from contextlib import contextmanager
+
 from django.core.exceptions import ObjectDoesNotExist
 from django.db import router, transaction
 from django.db.models import ProtectedError, RestrictedError
@@ -9,6 +11,7 @@ from rest_framework.response import Response
 from rest_framework.reverse import reverse
 
 from core.models import ObjectType
+from core.signals import clear_events
 from extras.models import ExportTemplate
 from netbox.api.serializers import BulkOperationSerializer
 from netbox.api.serializers.bulk import get_bulk_update_serializer_class
@@ -25,9 +28,43 @@ __all__ = (
     'ExportTemplatesMixin',
     'ObjectValidationMixin',
     'SequentialBulkCreatesMixin',
+    'discard_events_on_rollback',
 )
 
 
+@contextmanager
+def discard_events_on_rollback(sender, using=None):
+    """
+    Discard any queued events if the transaction wrapping this block is rolled back.
+
+    The change logging signal receivers queue events eagerly, as the payload for a deleted object
+    must be captured while that object and its related rows are still reachable. The queue is not
+    flushed to the events pipeline until after the response has been rendered, however, so events
+    queued for writes which were subsequently rolled back would otherwise still be dispatched,
+    firing webhooks and event rules for changes that were never committed.
+
+    Bulk operations need this because they provisionally write every valid object in a batch and
+    then roll the entire batch back if any one object failed. Single-object writes need it because
+    a write can be undone after it has been saved (for instance by the object-level permission
+    check in perform_create()/perform_update(), or by a signal receiver raising AbortRequest). The
+    UI's views send the same signal when they abandon a transaction.
+
+    Must be entered *inside* the transaction whose rollback it guards, so that the rollback flag is
+    still set when this block exits. Nesting is safe: the bulk actions guard the whole batch while
+    the per-object perform_*() calls they make guard each write, and clearing an already-empty
+    queue is a no-op.
+    """
+    try:
+        yield
+    except Exception:
+        # An exception escaping the block (e.g. AbortRequest raised by a signal receiver) rolls
+        # the transaction back just as an explicit set_rollback() does.
+        clear_events.send(sender=sender)
+        raise
+    if transaction.get_connection(using).needs_rollback:
+        clear_events.send(sender=sender)
+
+
 class BackgroundOperationMixin:
     """
     Enable optional background processing of REST API bulk write operations. When a write
@@ -169,7 +206,8 @@ class SequentialBulkCreatesMixin:
         # (e.g. rack space checks). Collect per-object errors instead of failing on the first.
         errors = []
         return_data = []
-        with transaction.atomic(using=router.db_for_write(self.queryset.model)):
+        using = router.db_for_write(self.queryset.model)
+        with transaction.atomic(using=using), discard_events_on_rollback(self, using=using):
             if not isinstance(request.data, list):
                 # Creating a single object
                 return super().create(request, *args, **kwargs)
@@ -273,7 +311,8 @@ class BulkUpdateModelMixin:
     def perform_bulk_update(self, objects, update_data, partial):
         updated_pks = []
         errors = []
-        with transaction.atomic(using=router.db_for_write(self.queryset.model)):
+        using = router.db_for_write(self.queryset.model)
+        with transaction.atomic(using=using), discard_events_on_rollback(self, using=using):
             # Validate and save each object in turn so subsequent validations see the DB
             # state left by prior saves (e.g. two items renamed to the same name: the second
             # will fail validation rather than raising an integrity error on save).
@@ -365,7 +404,8 @@ class BulkDestroyModelMixin:
         changelog_messages = changelog_messages or {}
         errors = []
         total = 0
-        with transaction.atomic(using=router.db_for_write(self.queryset.model)):
+        using = router.db_for_write(self.queryset.model)
+        with transaction.atomic(using=using), discard_events_on_rollback(self, using=using):
             for obj in objects:
                 total += 1
                 if hasattr(obj, 'snapshot'):