Fixes #22978: Discard queued events when a REST API write is rolled back

The REST API's write paths are transactional, but the in-memory events queue
is not: change logging receivers queue events eagerly (for deletions in
pre_delete, before the row is removed), and event_tracking() flushes that queue
only after the response has been rendered. Where a rolled-back write is caught
and converted into a normal response — the ProtectedError/RestrictedError and
AbortRequest handlers in NetBoxModelViewSet.dispatch(), and the
ObjectDoesNotExist to PermissionDenied conversions in the perform_*() methods —
the flush therefore dispatched events for objects that were never created,
updated, or deleted.

Backport discard_events_on_rollback() from the feature branch and enter it
inside the transaction guarding each API write: perform_create(),
perform_update(), and perform_destroy(); the SequentialBulkCreatesMixin,
BulkUpdateModelMixin, and BulkDestroyModelMixin bulk actions; and
AvailableObjectsView.post(). The context manager sends clear_events on the way
out whenever the wrapped transaction is rolled back, whether by an exception
escaping the block or by an explicit set_rollback(), which also covers the
exceptions that reach dispatch(). This mirrors what the UI views now do (#22934).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

netbox/extras/tests/test_event_rules.py

@@ -8,7 +8,7 @@ from unittest.mock import Mock, patch
 import django_rq
 from django.conf import settings
 from django.http import HttpResponse
-from django.test import RequestFactory, TestCase, tag
+from django.test import RequestFactory, TestCase, override_settings, tag
 from django.urls import reverse
 from PIL import Image
 from requests import Session
@@ -17,16 +17,19 @@ from rest_framework import status
 from core.choices import JobNotificationChoices, 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, serialize_for_event
 from extras.models import EventRule, Notification, Script, ScriptModule, Tag, Webhook
 from extras.scripts import Script as ScriptBase
 from extras.signals import process_job_end_event_rules
 from extras.webhooks import generate_signature, send_webhook
+from ipam.choices import IPAddressStatusChoices
+from ipam.models import IPAddress, Prefix
 from netbox.context_managers import event_tracking
-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
 
 
@@ -167,6 +170,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
@@ -218,6 +247,72 @@ 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_available_objects_create_rollback_discards_events(self):
+        """
+        Check that creating an object via an available-objects endpoint (e.g. available-ips) queues
+        no background task when the object-level permission check rolls the transaction back.
+        """
+        prefix = Prefix.objects.create(prefix='192.0.2.0/24')
+
+        event_rule = EventRule.objects.get(name='Event Rule 1')
+        event_rule.object_types.set([ObjectType.objects.get_for_model(IPAddress)])
+
+        # Permit the creation of active IP addresses 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': IPAddressStatusChoices.STATUS_ACTIVE},
+        )
+        obj_perm.save()
+        obj_perm.users.add(self.user)
+        obj_perm.object_types.add(ObjectType.objects.get_for_model(IPAddress))
+        self.add_permissions('ipam.view_prefix')
+
+        url = reverse('ipam-api:prefix-available-ips', kwargs={'pk': prefix.pk})
+        data = {'status': IPAddressStatusChoices.STATUS_RESERVED}
+        with disable_warnings('django.request'):
+            response = self.client.post(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN)
+        self.assertEqual(IPAddress.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.
@@ -252,6 +347,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
@@ -309,6 +435,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.
@@ -333,6 +491,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
@@ -367,6 +554,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/ipam/api/views.py

@@ -19,7 +19,7 @@ from ipam import filtersets
 from ipam.models import *
 from ipam.utils import get_next_available_prefix
 from netbox.api.viewsets import NetBoxModelViewSet
-from netbox.api.viewsets.mixins import ObjectValidationMixin
+from netbox.api.viewsets.mixins import ObjectValidationMixin, discard_events_on_rollback
 from netbox.config import get_config
 from netbox.constants import ADVISORY_LOCK_KEYS
 from utilities.api import get_serializer_for_model
@@ -295,8 +295,9 @@ class AvailableObjectsView(ObjectValidationMixin, APIView):
                 return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
 
             # Create the new IP address(es)
+            using = router.db_for_write(self.queryset.model)
             try:
-                with transaction.atomic(using=router.db_for_write(self.queryset.model)):
+                with transaction.atomic(using=using), discard_events_on_rollback(self, using=using):
                     created = serializer.save()
                     self._validate_objects(created)
             except ObjectDoesNotExist:

netbox/netbox/api/viewsets/__init__.py

@@ -260,8 +260,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:
@@ -299,8 +300,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):
@@ -333,8 +335,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.http import Http404
@@ -5,6 +7,7 @@ from rest_framework import status
 from rest_framework.response import Response
 
 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
@@ -16,9 +19,48 @@ __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.
+
+    Note that this discards the entire request's queue, not only the events queued within the
+    guarded block. Nesting is therefore safe only because every rollback guarded here aborts the
+    whole request, making the two equivalent: the bulk actions guard the whole batch while the
+    per-object perform_*() calls they make guard each write, and a failure in either case abandons
+    the request. Do not use this in a loop which catches a per-object failure and continues, as
+    the events for objects which were successfully written would be discarded as well.
+    """
+    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 CustomFieldsMixin:
     """
     For models which support custom fields, populate the `custom_fields` context.
@@ -61,7 +103,8 @@ class SequentialBulkCreatesMixin:
     appropriately.
     """
     def create(self, request, *args, **kwargs):
-        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)
@@ -122,7 +165,8 @@ class BulkUpdateModelMixin:
 
     def perform_bulk_update(self, objects, update_data, partial):
         updated_pks = []
-        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:
                 data = update_data.get(obj.id)
                 if hasattr(obj, 'snapshot'):
@@ -185,7 +229,8 @@ class BulkDestroyModelMixin:
 
     def perform_bulk_destroy(self, objects, changelog_messages=None):
         changelog_messages = changelog_messages or {}
-        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:
                 if hasattr(obj, 'snapshot'):
                     obj.snapshot()