perf(plugin): bulk ObjectChange writes via thread-local + bulk_create

NetBox's core.signals.handle_changed_object fires ObjectChange.save()
once per post_save / m2m_changed event. For a multi-thousand-entity
batch the per-instance INSERT becomes the dominant write cost.

Install a module-level monkey-patch on ObjectChange.save that, when a
thread-local flag is active, buffers the instance instead of writing
it. The deferred_changelog() context manager yields a _Deferred
accumulator with commit_pending() / rollback_pending() / flush()
methods:

* On success of a changeset, the caller calls commit_pending(): the
  changeset's pending audit-log rows promote into the batch buffer.
* On failure (ChangeSetException -> savepoint rollback), the caller
  calls rollback_pending(): the failed changeset's audit rows are
  discarded so they cannot leak past the rollback.
* At batch end, flush() drains the batch buffer with one
  ObjectChange.objects.bulk_create(buffer, batch_size=500).

m2m merge dedup is preserved by a per-window in-memory index keyed on
(content_type_id, object_id, request_id): a follow-up m2m_changed fire
collapses onto the prior buffered row instead of producing a duplicate
audit entry.

ApplyChangeSetView and BulkApplyView both wrap their work with
deferred_changelog(), so the batch's audit-log rows commit together
with the batch's data writes (per-batch atomicity to readers).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: ldrozdz93

netbox_diode_plugin/api/deferred_changelog.py

@@ -0,0 +1,196 @@
+#!/usr/bin/env python
+# Copyright 2026 NetBox Labs, Inc.
+"""
+Deferred ObjectChange writes for diode apply paths.
+
+PR 3 of BULK-ORM Sprint 1, with the V3 atomicity fix layered on top.
+
+NetBox's ``core.signals.handle_changed_object`` fires per-instance on
+``post_save`` / ``m2m_changed`` and persists each ``ObjectChange`` row
+with its own INSERT. For a multi-thousand-entity changeset that's the
+dominant write cost.
+
+This module installs a module-level monkey-patch on
+``core.models.ObjectChange.save`` that, when a thread-local flag is
+active, buffers the instance instead of writing it. The
+``deferred_changelog()`` context manager activates the flag and yields
+a ``_Deferred`` accumulator with three methods:
+
+* ``commit_pending()`` — promote the per-changeset *pending* buffer
+  into the cross-changeset *batch* buffer. Called after a changeset's
+  inner ``transaction.atomic()`` block completes successfully.
+* ``rollback_pending()`` — discard the per-changeset *pending* buffer.
+  Called when a changeset's inner atomic block raises and its writes
+  are rolled back. Without this the buffered ``ObjectChange`` rows
+  would still get flushed, leaking audit-log entries pointing at
+  rolled-back data.
+* ``flush()`` — issue ``ObjectChange.objects.bulk_create()`` for the
+  accumulated batch buffer. Caller invokes this once at the end of
+  the loop, **inside** the outer ``transaction.atomic()`` so a flush
+  failure rolls back the whole batch.
+
+m2m merge dedup is preserved by ``(content_type_id, object_id,
+request_id)`` keys spanning both buffers: a follow-up m2m_changed
+fire collapses onto whichever buffer holds the prior row, yielding
+a single audit row per object regardless of when within the batch
+the dedup happens.
+
+Nested usage is a no-op (outer wins).
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from contextlib import contextmanager
+
+from core.models import ObjectChange
+
+logger = logging.getLogger(__name__)
+
+_state = threading.local()
+_BATCH_SIZE = 500
+
+
+def _is_active() -> bool:
+    return getattr(_state, "active", False)
+
+
+def _get_deferred():
+    return getattr(_state, "deferred", None)
+
+
+class _Deferred:
+    """Per-window accumulator for buffered ``ObjectChange`` writes."""
+
+    def __init__(self) -> None:
+        self.pending_buffer: list[ObjectChange] = []
+        self.pending_index: dict[tuple, int] = {}
+        self.batch_buffer: list[ObjectChange] = []
+        self.batch_index: dict[tuple, int] = {}
+
+    def _add(self, obj: ObjectChange) -> None:
+        """Buffer ``obj`` into pending; merge onto pending or batch on dedup."""
+        key = (
+            obj.changed_object_type_id,
+            obj.changed_object_id,
+            str(obj.request_id) if obj.request_id is not None else None,
+        )
+        pos = self.pending_index.get(key)
+        if pos is not None:
+            self.pending_buffer[pos].postchange_data = obj.postchange_data
+            return
+        batch_pos = self.batch_index.get(key)
+        if batch_pos is not None:
+            self.batch_buffer[batch_pos].postchange_data = obj.postchange_data
+            return
+        self.pending_index[key] = len(self.pending_buffer)
+        self.pending_buffer.append(obj)
+
+    def commit_pending(self) -> None:
+        """Promote pending entries onto the batch buffer; clear pending."""
+        for key, pos in self.pending_index.items():
+            obj = self.pending_buffer[pos]
+            existing = self.batch_index.get(key)
+            if existing is not None:
+                self.batch_buffer[existing].postchange_data = obj.postchange_data
+            else:
+                self.batch_index[key] = len(self.batch_buffer)
+                self.batch_buffer.append(obj)
+        self.pending_buffer = []
+        self.pending_index = {}
+
+    def rollback_pending(self) -> None:
+        """Discard the pending buffer (e.g. after a savepoint rollback)."""
+        self.pending_buffer = []
+        self.pending_index = {}
+
+    def flush(self) -> None:
+        """Drain pending into batch, then bulk-insert and clear the batch buffer.
+
+        Calling ``flush()`` without first calling ``commit_pending()`` is the
+        legacy single-window pattern (``with deferred_changelog(): ...`` with
+        no per-changeset accumulator). It commits any leftover pending entries
+        before issuing the INSERT so callers that don't use the V3 lifecycle
+        API still get a single bulk_create.
+        """
+        if self.pending_buffer:
+            self.commit_pending()
+        if self.batch_buffer:
+            ObjectChange.objects.bulk_create(self.batch_buffer, batch_size=_BATCH_SIZE)
+        self.batch_buffer = []
+        self.batch_index = {}
+
+
+_original_save = ObjectChange.save
+
+
+def _patched_save(self, *args, **kwargs):
+    """ObjectChange.save replacement: buffers when deferred_changelog() is active."""
+    if not _is_active():
+        return _original_save(self, *args, **kwargs)
+
+    deferred = _get_deferred()
+    if deferred is None:
+        return _original_save(self, *args, **kwargs)
+
+    if self.pk is not None:
+        # Already in DB (e.g. m2m post_clear updating an externally-saved
+        # change row). Fall back to the real save for correctness.
+        return _original_save(self, *args, **kwargs)
+
+    if not self.user_name and getattr(self, "user_id", None):
+        self.user_name = self.user.username
+    if not self.object_repr:
+        try:
+            self.object_repr = str(self.changed_object) if self.changed_object_id else ""
+        except Exception:
+            self.object_repr = ""
+
+    deferred._add(self)
+    return None
+
+
+# Install the monkey-patch at import time; views.py imports this module.
+ObjectChange.save = _patched_save
+
+
+@contextmanager
+def deferred_changelog():
+    """
+    Activate deferred ``ObjectChange`` persistence; yield a ``_Deferred``.
+
+    Two usage modes are supported:
+
+    * Legacy (single-window) — ``with deferred_changelog(): ...``. All saves
+      go to the pending buffer; on clean context exit the manager auto-calls
+      ``flush()``, draining pending into batch and issuing one
+      ``bulk_create``. If the body raises, no flush happens, so an unhandled
+      exception cannot leak audit-log rows.
+
+    * V3 (per-changeset accumulator) — ``with deferred_changelog() as defc:``
+      with explicit ``defc.commit_pending()`` / ``defc.rollback_pending()``
+      after each savepoint and ``defc.flush()`` once at the end inside the
+      outer ``transaction.atomic()``. The auto-flush at clean context exit
+      is a no-op because ``flush()`` already cleared the buffers.
+
+    Nested invocation is a no-op (outer wins).
+    """
+    if _is_active():
+        yield _get_deferred()
+        return
+
+    deferred = _Deferred()
+    _state.active = True
+    _state.deferred = deferred
+    raised = False
+    try:
+        yield deferred
+    except Exception:
+        raised = True
+        raise
+    finally:
+        _state.active = False
+        _state.deferred = None
+        if not raised:
+            deferred.flush()

netbox_diode_plugin/api/views.py

@@ -17,6 +17,7 @@
     ChangeSetException,
     ChangeSetResult,
 )
+from .deferred_changelog import deferred_changelog
 from .differ import enter_prechange_cache, exit_prechange_cache, generate_changeset
 from .matcher import enter_request_obj_cache, exit_request_obj_cache
 from .permissions import (
@@ -317,7 +318,13 @@ def post(self, request, *args, **kwargs):
 
     def _post(self, request, *args, **kwargs):
         change_set = ChangeSet.from_dict(request.data)
-        result = _apply_one_changeset(change_set, request)
+        with deferred_changelog() as defc:
+            result = _apply_one_changeset(change_set, request)
+            if result.errors:
+                defc.rollback_pending()
+            else:
+                defc.commit_pending()
+            defc.flush()
         return Response(result.to_dict(), status=result.get_status_code())
 
 
@@ -358,23 +365,31 @@ def _post(self, request, *args, **kwargs):
             raise ValidationError({"change_sets": ["change_sets must not be empty"]})
 
         results = []
-        for entry in change_sets:
-            if not isinstance(entry, dict):
-                results.append(
-                    ChangeSetResult(
-                        errors={"request": {"change_set": ["change_set must be an object"]}}
+        with deferred_changelog() as defc:
+            for entry in change_sets:
+                if not isinstance(entry, dict):
+                    results.append(
+                        ChangeSetResult(
+                            errors={"request": {"change_set": ["change_set must be an object"]}}
+                        ).to_dict()
+                    )
+                    continue
+                try:
+                    change_set = ChangeSet.from_dict(entry)
+                    cs_result = _apply_one_changeset(change_set, request)
+                    if cs_result.errors:
+                        defc.rollback_pending()
+                    else:
+                        defc.commit_pending()
+                    result = cs_result.to_dict()
+                except Exception as e:
+                    logger.error(f"Error parsing batch entry: {e}")
+                    defc.rollback_pending()
+                    result = ChangeSetResult(
+                        errors={"request": {"change_set": [f"invalid change_set: {e}"]}}
                     ).to_dict()
-                )
-                continue
-            try:
-                change_set = ChangeSet.from_dict(entry)
-                result = _apply_one_changeset(change_set, request).to_dict()
-            except Exception as e:
-                logger.error(f"Error parsing batch entry: {e}")
-                result = ChangeSetResult(
-                    errors={"request": {"change_set": [f"invalid change_set: {e}"]}}
-                ).to_dict()
-            results.append(result)
+                results.append(result)
+            defc.flush()
 
         http_status = (
             status.HTTP_207_MULTI_STATUS

netbox_diode_plugin/tests/v4.5.x/tests/test_deferred_changelog.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python
+# Copyright 2026 NetBox Labs, Inc.
+"""Diode NetBox Plugin - PR 3 deferred changelog tests."""
+
+import uuid
+
+from core.models import ObjectChange
+from dcim.models import Site
+from django.contrib.contenttypes.models import ContentType
+from django.db import connection
+from django.test import RequestFactory, TestCase
+from django.test.utils import CaptureQueriesContext
+from netbox.context import current_request
+
+from netbox_diode_plugin.api.deferred_changelog import deferred_changelog
+from netbox_diode_plugin.plugin_config import get_diode_user
+
+
+def _make_request():
+    """Build a minimal request with a fresh request_id, attached to current_request."""
+    rf = RequestFactory()
+    req = rf.post("/x")
+    req.id = uuid.uuid4()
+    req.user = get_diode_user()
+    return req
+
+
+class DeferredChangelogTestCase(TestCase):
+    """Verifies deferred_changelog (PR 3)."""
+
+    def test_n_changes_one_bulk_insert(self):
+        """N saves inside the context produce ONE bulk INSERT, not N."""
+        req = _make_request()
+        token = current_request.set(req)
+        try:
+            ObjectChange.objects.all().delete()
+            with CaptureQueriesContext(connection) as ctx, deferred_changelog():
+                for i in range(5):
+                    Site.objects.create(name=f"S-{i}", slug=f"s-{i}")
+
+            insert_stmts = [
+                q for q in ctx.captured_queries
+                if "INSERT INTO" in q["sql"] and '"core_objectchange"' in q["sql"]
+            ]
+            assert len(insert_stmts) == 1, [q["sql"] for q in insert_stmts]
+            assert ObjectChange.objects.count() == 5
+        finally:
+            current_request.reset(token)
+
+    def test_audit_row_content_matches_baseline(self):
+        """Buffered audit rows carry the same fields as the per-save baseline."""
+        req = _make_request()
+        token = current_request.set(req)
+        try:
+            ObjectChange.objects.all().delete()
+            with deferred_changelog():
+                Site.objects.create(name="Deferred S", slug="deferred-s")
+            deferred_row = ObjectChange.objects.get(object_repr="Deferred S")
+
+            ObjectChange.objects.all().delete()
+            req2 = _make_request()
+            token2 = current_request.set(req2)
+            try:
+                Site.objects.create(name="Baseline S", slug="baseline-s")
+            finally:
+                current_request.reset(token2)
+            baseline_row = ObjectChange.objects.get(object_repr="Baseline S")
+
+            for field in ("action", "user_name", "changed_object_type_id",
+                          "postchange_data"):
+                assert getattr(deferred_row, field) == getattr(baseline_row, field), field
+        finally:
+            current_request.reset(token)
+
+    def test_nested_context_is_noop(self):
+        """An inner deferred_changelog must not flush — outer wins."""
+        req = _make_request()
+        token = current_request.set(req)
+        try:
+            ObjectChange.objects.all().delete()
+            with deferred_changelog():
+                with deferred_changelog():
+                    Site.objects.create(name="N-1", slug="n-1")
+                # If inner had flushed, this row would already exist.
+                assert ObjectChange.objects.filter(object_repr="N-1").count() == 0
+                Site.objects.create(name="N-2", slug="n-2")
+
+            assert ObjectChange.objects.filter(object_repr__in=["N-1", "N-2"]).count() == 2
+        finally:
+            current_request.reset(token)
+
+    def test_m2m_merge_dedup_single_row(self):
+        """An UPDATE that re-fires via m2m_changed merges into one buffered row."""
+        from extras.models import Tag
+
+        tag = Tag.objects.create(name="net", slug="net")
+        site = Site.objects.create(name="MM", slug="mm")  # outside the window
+
+        req = _make_request()
+        token = current_request.set(req)
+        try:
+            ObjectChange.objects.all().delete()
+            with deferred_changelog():
+                site.description = "updated"
+                site.save()  # post_save → ObjectChange #1
+                site.tags.set([tag])  # m2m_changed → would normally update the prior row
+
+            ct = ContentType.objects.get_for_model(Site)
+            rows = list(ObjectChange.objects.filter(
+                changed_object_type=ct,
+                changed_object_id=site.pk,
+                request_id=req.id,
+            ))
+            assert len(rows) == 1, [r.postchange_data for r in rows]
+            assert rows[0].postchange_data.get("tags") == [tag.id]
+        finally:
+            current_request.reset(token)