#21025: Optimize rendering of config context data (#22294)

* #21025: WIP

* Fixes #22357: Remove unused `local_context_data` field from dcim.Module (#22364)

* Add partial index for checking null CC data

* Ensure the data returned by get_config_context() is safe for mutation

* Implement selective backup queryset annotation to avoid n+1 overhead on cold cache

* Fix migration conflict

* Replace MPTT with Ltree per #21418

docs/features/context-data.md

@@ -90,3 +90,14 @@ Devices and virtual machines may also have a local context data defined. This lo
 A [config context profile](../models/extras/configcontextprofile.md) provides an organizational grouping for related config contexts and may optionally enforce a [JSON schema](https://json-schema.org/) describing the shape of their data. When a profile is assigned to a config context, NetBox validates the context's data against the profile's schema on save and rejects any context that fails validation. This makes it possible to constrain which keys may appear in a context, require certain keys to be present, or limit values to a defined enumeration — guarding against typos and drift as contexts proliferate.
 
 A profile's schema may be authored directly in NetBox or populated from an external [data source](../models/core/datasource.md), enabling teams to maintain schemas alongside the code or configurations that consume them.
+
+## Pre-rendered Caching
+
+!!! info "New in NetBox v4.7"
+
+NetBox pre-renders each device's and virtual machine's merged context data and stores it on the object itself, so most reads can return the result without recomputing the full set of applicable contexts. The cache is initially populated during upgrade (the upgrade script runs the `rebuild_config_context_cache` management command) and is thereafter kept current automatically: whenever an upstream change is detected — a config context being created, modified, or deleted; a device/VM's scope-relevant attribute changing (site, role, tenant, tags, cluster, etc.); or a related object being re-routed in a way that changes which contexts apply — NetBox marks the affected caches invalid and enqueues a non-blocking [background job](./background-jobs.md) to repopulate them.
+
+During the brief window between invalidation and re-render, requests for the affected object's config context fall back to the original on-demand rendering path, so the data returned is always correct — never stale — but may be slightly slower during that window. Once the background job completes, reads are served from the cache.
+
+!!! note
+    The pre-rendered cache supersedes the previous `?exclude=config_context` REST API query parameter. Config context data is now always returned for devices and virtual machines, and the parameter is silently ignored.

netbox/dcim/api/serializers_/devices.py

@@ -27,7 +27,6 @@ from .virtualchassis import VirtualChassisSerializer
 
 __all__ = (
     'DeviceSerializer',
-    'DeviceWithConfigContextSerializer',
     'MACAddressSerializer',
     'ModuleSerializer',
     'VirtualDeviceContextSerializer',
@@ -87,6 +86,7 @@ class DeviceSerializer(PrimaryModelSerializer):
     virtual_chassis = VirtualChassisSerializer(nested=True, required=False, allow_null=True, default=None)
     vc_position = serializers.IntegerField(allow_null=True, max_value=255, min_value=0, default=None)
     config_template = ConfigTemplateSerializer(nested=True, required=False, allow_null=True, default=None)
+    config_context = serializers.SerializerMethodField(read_only=True)
 
     # Counter fields
     console_port_count = serializers.IntegerField(read_only=True)
@@ -106,10 +106,10 @@ class DeviceSerializer(PrimaryModelSerializer):
             'id', 'url', 'display_url', 'display', 'name', 'device_type', 'role', 'tenant', 'platform', 'serial',
             'asset_tag', 'site', 'location', 'rack', 'position', 'face', 'latitude', 'longitude', 'parent_device',
             'status', 'airflow', 'primary_ip', 'primary_ip4', 'primary_ip6', 'oob_ip', 'cluster', 'virtual_chassis',
-            'vc_position', 'vc_priority', 'description', 'owner', 'comments', 'config_template', 'local_context_data',
-            'tags', 'custom_fields', 'created', 'last_updated', 'console_port_count', 'console_server_port_count',
-            'power_port_count', 'power_outlet_count', 'interface_count', 'front_port_count', 'rear_port_count',
-            'device_bay_count', 'module_bay_count', 'inventory_item_count',
+            'vc_position', 'vc_priority', 'description', 'owner', 'comments', 'config_template', 'config_context',
+            'local_context_data', 'tags', 'custom_fields', 'created', 'last_updated', 'console_port_count',
+            'console_server_port_count', 'power_port_count', 'power_outlet_count', 'interface_count',
+            'front_port_count', 'rear_port_count', 'device_bay_count', 'module_bay_count', 'inventory_item_count',
         ]
         brief_fields = ('id', 'url', 'display', 'name', 'description')
 
@@ -124,21 +124,6 @@ class DeviceSerializer(PrimaryModelSerializer):
         data['device_bay'] = NestedDeviceBaySerializer(instance=device_bay, context=context).data
         return data
 
-
-class DeviceWithConfigContextSerializer(DeviceSerializer):
-    config_context = serializers.SerializerMethodField(read_only=True, allow_null=True)
-
-    class Meta(DeviceSerializer.Meta):
-        fields = [
-            'id', 'url', 'display_url', 'display', 'name', 'device_type', 'role', 'tenant', 'platform', 'serial',
-            'asset_tag', 'site', 'location', 'rack', 'position', 'face', 'latitude', 'longitude', 'parent_device',
-            'status', 'airflow', 'primary_ip', 'primary_ip4', 'primary_ip6', 'oob_ip', 'cluster', 'virtual_chassis',
-            'vc_position', 'vc_priority', 'description', 'owner', 'comments', 'config_template', 'config_context',
-            'local_context_data', 'tags', 'custom_fields', 'created', 'last_updated', 'console_port_count',
-            'console_server_port_count', 'power_port_count', 'power_outlet_count', 'interface_count',
-            'front_port_count', 'rear_port_count', 'device_bay_count', 'module_bay_count', 'inventory_item_count',
-        ]
-
     @extend_schema_field(serializers.JSONField(allow_null=True))
     def get_config_context(self, obj):
         return obj.get_config_context()

netbox/dcim/api/views.py

@@ -410,35 +410,15 @@ class PlatformViewSet(NetBoxModelViewSet):
 # Devices/modules
 #
 
-class DeviceViewSet(
-    SequentialBulkCreatesMixin,
-    ConfigContextQuerySetMixin,
-    RenderConfigMixin,
-    NetBoxModelViewSet
-):
+class DeviceViewSet(SequentialBulkCreatesMixin, ConfigContextQuerySetMixin, RenderConfigMixin, NetBoxModelViewSet):
     queryset = Device.objects.prefetch_related(
         'device_type__manufacturer',  # Referenced by Device.__str__() for unnamed devices
         'parent_bay',  # Referenced by DeviceSerializer.get_parent_device()
     )
+    serializer_class = serializers.DeviceSerializer
     filterset_class = filtersets.DeviceFilterSet
     pagination_class = StripCountAnnotationsPaginator
 
-    def get_serializer_class(self):
-        """
-        Select the specific serializer based on the request context.
-
-        If the `brief` query param equates to True, return the NestedDeviceSerializer
-
-        If the `exclude` query param includes `config_context` as a value, return the DeviceSerializer
-
-        Else, return the DeviceWithConfigContextSerializer
-        """
-        request = self.get_serializer_context()['request']
-        if self.brief or 'config_context' in request.query_params.get('exclude', []):
-            return serializers.DeviceSerializer
-
-        return serializers.DeviceWithConfigContextSerializer
-
 
 class VirtualDeviceContextViewSet(NetBoxModelViewSet):
     queryset = VirtualDeviceContext.objects.all()

netbox/dcim/migrations/0239_device__config_context_data.py

@@ -0,0 +1,26 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ('dcim', '0238_ltree_paths'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='device',
+            name='_config_context_data',
+            field=models.JSONField(blank=True, editable=False, null=True),
+        ),
+        migrations.AddField(
+            model_name='device',
+            name='_config_context_generation',
+            field=models.PositiveBigIntegerField(default=0, editable=False),
+        ),
+        migrations.AddIndex(
+            model_name='device',
+            index=models.Index(
+                condition=models.Q(('_config_context_data__isnull', True)), fields=['id'], name='dcim_device_cc_null'
+            ),
+        ),
+    ]

netbox/dcim/models/devices.py

@@ -751,6 +751,15 @@ class Device(
         ordering = ('name', 'pk')  # Name may be null
         indexes = (
             models.Index(fields=('name', 'id')),  # Default ordering
+            # Partial index supporting the background renderer's scan for objects whose config
+            # context cache has been invalidated (see extras.jobs.RenderConfigContextJob and
+            # extras.cache). Indexing only the NULL-cache rows keeps that lookup cheap on large
+            # tables where the steady state is for nearly every row to be populated.
+            models.Index(
+                fields=('id',),
+                condition=Q(_config_context_data__isnull=True),
+                name='dcim_device_cc_null',
+            ),
         )
         constraints = (
             models.UniqueConstraint(

netbox/dcim/tests/test_api.py

@@ -1814,16 +1814,6 @@ class DeviceTestCase(APIViewTestCases.APIViewTestCase):
 
         self.assertEqual(response.data['results'][0].get('config_context', {}).get('A'), 1)
 
-    def test_config_context_excluded(self):
-        """
-        Check that config context data can be excluded by passing ?exclude=config_context.
-        """
-        self.add_permissions('dcim.view_device')
-        url = reverse('dcim-api:device-list') + '?exclude=config_context'
-        response = self.client.get(url, **self.header)
-
-        self.assertFalse('config_context' in response.data['results'][0])
-
     def test_unique_name_per_site_constraint(self):
         """
         Check that creating a device with a duplicate name within a site fails.
@@ -1975,7 +1965,7 @@ class DeviceTestCase(APIViewTestCases.APIViewTestCase):
 
         self.add_permissions('dcim.view_device', 'ipam.view_ipaddress')
         response = self.client.get(
-            f'{self._get_detail_url(device)}?exclude=config_context',
+            self._get_detail_url(device),
             **self.header,
         )
         self.assertHttpStatus(response, status.HTTP_200_OK)
@@ -2003,7 +1993,7 @@ class DeviceTestCase(APIViewTestCases.APIViewTestCase):
 
         self.add_permissions('dcim.view_device', 'ipam.view_ipaddress')
         response = self.client.get(
-            f'{self._get_detail_url(device)}?exclude=config_context',
+            self._get_detail_url(device),
             **self.header,
         )
         self.assertHttpStatus(response, status.HTTP_200_OK)

netbox/dcim/views.py

@@ -2860,7 +2860,7 @@ class DeviceInventoryView(DeviceComponentsView):
 
 @register_model_view(Device, 'configcontext', path='config-context')
 class DeviceConfigContextView(ObjectConfigContextView):
-    queryset = Device.objects.annotate_config_context_data()
+    queryset = Device.objects.all()
     base_template = 'dcim/device/base.html'
     tab = ViewTab(
         label=_('Config Context'),

netbox/extras/api/mixins.py

@@ -20,24 +20,22 @@ __all__ = (
 
 class ConfigContextQuerySetMixin:
     """
-    Used by views that work with config context models (device and virtual machine).
-    Provides a get_queryset() method which deals with adding the config context
-    data annotation or not.
+    Used by viewsets for config context models (Device, VirtualMachine).
+
+    For non-brief requests, annotates the queryset so that config context data is computed in a
+    single query for any object whose pre-rendered cache (`_config_context_data`) has been
+    invalidated (NULL). Objects with a warm cache are served directly from it by
+    ConfigContextModel.get_config_context() and incur no subquery — PostgreSQL short-circuits the
+    CASE, so the correlated aggregation runs only for the invalidated rows. This avoids the
+    per-object fallback query that would otherwise occur when listing objects with cold caches
+    (e.g. immediately following an upgrade or a broad invalidation).
     """
     def get_queryset(self):
-        """
-        Build the proper queryset based on the request context
-
-        If the `brief` query param equates to True or the `exclude` query param
-        includes `config_context` as a value, return the base queryset.
-
-        Else, return the queryset annotated with config context data
-        """
         queryset = super().get_queryset()
-        request = self.get_serializer_context()['request']
-        if self.brief or 'config_context' in request.query_params.get('exclude', []):
+        # Brief responses omit config_context entirely, so the annotation would be pure overhead.
+        if self.brief:
             return queryset
-        return queryset.annotate_config_context_data()
+        return queryset.annotate_config_context_data(only_invalidated=True)
 
 
 class ConfigTemplateRenderMixin:

netbox/extras/cache.py

@@ -0,0 +1,144 @@
+"""
+Invalidation helpers for the pre-rendered config-context cache on Device and VirtualMachine.
+
+Every signal handler in extras/signals.py that needs to invalidate cached config-context data
+funnels through this module, so the synchronous NULL-out and background job enqueue are
+expressed in exactly one place.
+"""
+from django.apps import apps
+from django.db import transaction
+from django.db.models import F, Q
+
+from dcim.models import Device
+from extras.jobs import RenderConfigContextJob
+from extras.models.tags import TaggedItem
+from virtualization.models import VirtualMachine
+
+
+def invalidate_config_context_for_objects(model_label, pks):
+    """
+    Synchronously NULL the `_config_context_data` cache on the given objects (bumping the
+    generation counter so an in-flight render can't overwrite the invalidation), then enqueue a
+    background job to repopulate them once the surrounding transaction commits.
+
+    Args:
+        model_label: 'dcim.device' or 'virtualization.virtualmachine'.
+        pks: Any iterable of object PKs (queryset, list, set, generator). An empty iterable is a no-op.
+    """
+    pks = list(pks)
+    if not pks:
+        return
+
+    Model = apps.get_model(model_label)
+    updated = Model.objects.filter(pk__in=pks).update(
+        _config_context_data=None,
+        _config_context_generation=F('_config_context_generation') + 1,
+    )
+    if not updated:
+        return
+
+    # Defer enqueue until after the current transaction commits, so the background worker doesn't
+    # try to read uncommitted state. transaction.on_commit() is a no-op outside a transaction,
+    # in which case the callback runs immediately.
+    #
+    # We deliberately enqueue a *parameterless* sweep (model_label=None, pks=None) that re-renders
+    # every object whose cache is currently NULL, rather than a job scoped to these specific PKs.
+    # JobRunner.enqueue_once() coalesces against any already-pending job of the same class (object_id
+    # is NULL for all of these, so get_jobs(None) matches them all); a job carrying a specific PKs
+    # list would therefore be silently dropped whenever another invalidation already had one enqueued,
+    # leaving those objects NULLed but never re-rendered. A global NULL-sweep makes coalescing correct:
+    # whichever sweep runs next picks up *all* outstanding NULL caches across both models. (Reads
+    # remain correct in the interim because get_config_context() renders on demand when the cache is
+    # NULL; the sweep only restores the pre-rendered fast path.)
+    transaction.on_commit(
+        lambda: RenderConfigContextJob.enqueue_once(instance=None)
+    )
+
+
+def invalidate_config_context_for_configcontext(configcontext):
+    """
+    Invalidate caches for all objects currently in scope for the given ConfigContext.
+    """
+    for queryset in configcontext.get_affected_objects():
+        invalidate_config_context_for_objects(
+            queryset.model._meta.label_lower,
+            queryset.values_list('pk', flat=True),
+        )
+
+
+def invalidate_for_scope_delta(scope_field, scope_pks):
+    """
+    Invalidate the cache of every Device/VirtualMachine that is matchable via the given scope
+    items, regardless of which ConfigContext those items belong to. Used when items are removed
+    from a ConfigContext scope (so we don't know the new affected set under that scope, only the
+    items that used to extend it).
+
+    `scope_field` is the ConfigContext M2M attribute name ('sites', 'regions', 'tags', ...).
+    `scope_pks` is the iterable of PKs of scope items that were removed/cleared.
+    """
+    scope_pks = list(scope_pks or ())
+    if not scope_pks:
+        return
+
+    device_q = None
+    vm_q = None
+
+    # Nested (ltree) scopes: any device/VM whose corresponding attribute resolves into the subtree
+    # of any of the changed items (descendant-or-equal).
+    nested_attrs = {
+        'regions': ('dcim', 'Region', 'site__region__path'),
+        'site_groups': ('dcim', 'SiteGroup', 'site__group__path'),
+        'roles': ('dcim', 'DeviceRole', 'role__path'),
+        'platforms': ('dcim', 'Platform', 'platform__path'),
+        'locations': ('dcim', 'Location', 'location__path'),  # Devices only
+    }
+    direct_attrs = {
+        'sites': 'site__in',
+        'cluster_types': 'cluster__type__in',
+        'cluster_groups': 'cluster__group__in',
+        'clusters': 'cluster__in',
+        'tenant_groups': 'tenant__group__in',
+        'tenants': 'tenant__in',
+        'device_types': 'device_type__in',  # Devices only
+    }
+
+    if scope_field in nested_attrs:
+        app, model_name, object_path = nested_attrs[scope_field]
+        Model = apps.get_model(app, model_name)
+        subtree_q = Q()
+        for path in Model.objects.filter(pk__in=scope_pks).values_list('path', flat=True):
+            subtree_q |= Q(**{f'{object_path}__descendant_or_equal': path})
+        if not subtree_q:
+            return
+        device_q = subtree_q
+        if scope_field != 'locations':
+            vm_q = subtree_q
+    elif scope_field in direct_attrs:
+        attr_path = direct_attrs[scope_field]
+        device_q = Q(**{attr_path: scope_pks})
+        if scope_field != 'device_types':
+            vm_q = Q(**{attr_path: scope_pks})
+    elif scope_field == 'tags':
+        device_tagged = TaggedItem.objects.filter(
+            tag_id__in=scope_pks,
+            content_type__app_label='dcim',
+            content_type__model='device',
+        ).values_list('object_id', flat=True)
+        vm_tagged = TaggedItem.objects.filter(
+            tag_id__in=scope_pks,
+            content_type__app_label='virtualization',
+            content_type__model='virtualmachine',
+        ).values_list('object_id', flat=True)
+        device_q = Q(pk__in=device_tagged)
+        vm_q = Q(pk__in=vm_tagged)
+    else:
+        return
+
+    if device_q is not None:
+        invalidate_config_context_for_objects(
+            'dcim.device', Device.objects.filter(device_q).values_list('pk', flat=True)
+        )
+    if vm_q is not None:
+        invalidate_config_context_for_objects(
+            'virtualization.virtualmachine', VirtualMachine.objects.filter(vm_q).values_list('pk', flat=True)
+        )

netbox/extras/constants.py

@@ -201,3 +201,15 @@ LOG_LEVEL_RANK = {
     LogLevelChoices.LOG_WARNING: 3,
     LogLevelChoices.LOG_FAILURE: 4,
 }
+
+# Config context cache: fields whose modification on an object requires re-rendering its config
+# context cache, keyed by model label.
+CC_FIELDS_BY_MODEL = {
+    'dcim.device': (
+        'site_id', 'location_id', 'device_type_id', 'role_id', 'tenant_id', 'platform_id',
+        'cluster_id', 'local_context_data',
+    ),
+    'virtualization.virtualmachine': (
+        'site_id', 'cluster_id', 'tenant_id', 'platform_id', 'role_id', 'local_context_data',
+    ),
+}

netbox/extras/graphql/mixins.py

@@ -29,15 +29,20 @@ class ConfigContextMixin:
     def get_queryset(cls, queryset, info: Info, **kwargs):
         queryset = super().get_queryset(queryset, info, **kwargs)
 
-        # If `config_context` is requested, call annotate_config_context_data() on the queryset
+        # When `config_context` is requested, annotate the aggregated context data — but only for
+        # rows whose pre-rendered cache (`_config_context_data`) is invalidated (NULL). Warm rows
+        # are served from the cache by get_config_context() and skip the subquery entirely
+        # (PostgreSQL short-circuits the CASE), so resolving config_context across a list of objects
+        # with cold caches no longer incurs a per-object fallback query.
         selected = {f.name for f in info.selected_fields[0].selections}
         if 'config_context' in selected and hasattr(queryset, 'annotate_config_context_data'):
-            return queryset.annotate_config_context_data()
+            return queryset.annotate_config_context_data(only_invalidated=True)
 
         return queryset
 
-    # Ensure `local_context_data` is fetched when `config_context` is requested
-    @strawberry_django.field(only=['local_context_data'])
+    # Ensure both the pre-rendered cache and `local_context_data` are fetched when `config_context`
+    # is requested, so the warm-cache read path requires no additional queries.
+    @strawberry_django.field(only=['_config_context_data', 'local_context_data'])
     def config_context(self) -> strawberry.scalars.JSON:
         return self.get_config_context()
 

netbox/extras/jobs.py

@@ -2,6 +2,7 @@ import logging
 import traceback
 from contextlib import ExitStack
 
+from django.apps import apps
 from django.db import DEFAULT_DB_ALIAS, router, transaction
 from django.utils.translation import gettext as _
 
@@ -15,6 +16,97 @@ from utilities.exceptions import AbortScript, AbortTransaction
 
 from .utils import is_report
 
+RENDER_CONFIG_CONTEXT_CHUNK_SIZE = 500
+
+# Safety bound on the number of re-scan passes performed by RenderConfigContextJob.run() (see the
+# loop there). Each pass re-queries for NULL caches, so any finite burst of concurrent
+# invalidations is drained well within this limit; the cap only guards against an object whose
+# cache is being invalidated faster than it can be rendered (pathological, unbounded churn).
+RENDER_CONFIG_CONTEXT_MAX_PASSES = 100
+
+
+class RenderConfigContextJob(JobRunner):
+    """
+    Recompute the pre-rendered `_config_context_data` cache for a set of Devices or
+    VirtualMachines. Enqueued (coalesced) by the invalidation helpers in extras/cache.py whenever
+    an upstream change (ConfigContext, related object, or the object itself) NULLs a cache.
+
+    This is *not* a recurring system job: the initial post-upgrade population is handled by the
+    `rebuild_config_context_cache` management command, and steady-state freshness is maintained by
+    the invalidation signals.
+    """
+
+    class Meta:
+        name = 'Render config context'
+
+    def run(self, model_label=None, pks=None, **kwargs):
+        """
+        Args:
+            model_label: 'dcim.device' or 'virtualization.virtualmachine'. If None, both are processed.
+            pks: An iterable of object PKs to refresh. If None, refresh all objects whose cache is null.
+        """
+        labels = (model_label,) if model_label is not None else ('dcim.device', 'virtualization.virtualmachine')
+        pks = list(pks) if pks is not None else None
+
+        # Re-scan until a full pass renders nothing. An invalidation that commits while this job is
+        # already RUNNING coalesces into this job — JobRunner.enqueue_once() treats RUNNING as an
+        # enqueued state — so it will NOT schedule a follow-up job. If we rendered in a single pass,
+        # any cache NULLed after the iterator moved past its row (or after the pass for its model
+        # completed) would be left populated by no one, stranding it on the on-demand read path
+        # indefinitely. Looping until a pass finds no renderable NULL caches guarantees those late
+        # invalidations are picked up before this job finishes.
+        total = 0
+        for _pass in range(RENDER_CONFIG_CONTEXT_MAX_PASSES):
+            rendered = sum(self._render_for_model(label, pks=pks) for label in labels)
+            total += rendered
+            # No progress this pass means either nothing is NULL or the only NULL rows are churning
+            # under concurrent invalidation (each such invalidation enqueues its own follow-up), so
+            # there is nothing more for us to safely do.
+            if not rendered:
+                break
+        else:
+            # The loop ran every pass without ever rendering nothing, meaning caches are being
+            # invalidated about as fast as we can render them. This is pathological churn worth
+            # surfacing: each lingering invalidation enqueues its own follow-up job, so the caches
+            # are not stranded, but the sustained rate warrants investigation.
+            self.logger.warning(
+                f"Reached the maximum of {RENDER_CONFIG_CONTEXT_MAX_PASSES} render passes with caches "
+                f"still being invalidated; config context caches may be churning under sustained "
+                f"concurrent invalidation."
+            )
+
+        self.logger.info(f"Rendered config context for {total} object(s)")
+
+    def _render_for_model(self, model_label, pks):
+        """
+        Render and cache config context for every object of the given model whose cache is
+        currently NULL (optionally restricted to `pks`). Returns the number of objects written.
+        """
+        Model = apps.get_model(model_label)
+        qs = Model.objects.filter(_config_context_data__isnull=True)
+        if pks is not None:
+            qs = qs.filter(pk__in=list(pks))
+
+        # Annotate so each instance's render() uses the same aggregated subquery the on-demand
+        # path would use, avoiding N additional queries.
+        qs = qs.annotate_config_context_data()
+
+        rendered = 0
+        for obj in qs.iterator(chunk_size=RENDER_CONFIG_CONTEXT_CHUNK_SIZE):
+            # Capture the generation we rendered against, then write the result back only if no
+            # invalidation has bumped it in the meantime (compare-and-set). If a fresh invalidation
+            # won the race, the row stays NULL with a higher generation and the follow-up job it
+            # enqueued will re-render it — we never persist a stale value.
+            generation = obj._config_context_generation
+            data = obj.render_config_context()
+            updated = Model.objects.filter(
+                pk=obj.pk,
+                _config_context_generation=generation,
+            ).update(_config_context_data=data)
+            rendered += updated
+
+        return rendered
+
 
 class ScriptJob(JobRunner):
     """

netbox/extras/management/commands/rebuild_config_context_cache.py

@@ -0,0 +1,47 @@
+from django.apps import apps
+from django.core.management.base import BaseCommand
+
+from extras.jobs import RENDER_CONFIG_CONTEXT_CHUNK_SIZE
+
+MODELS = ('dcim.device', 'virtualization.virtualmachine')
+
+
+class Command(BaseCommand):
+    help = "Pre-render and cache config context data for all devices and virtual machines"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--force', action='store_true',
+            help="Re-render every object, including those whose cache is already populated"
+        )
+
+    def handle(self, *args, **options):
+        for model_label in MODELS:
+            Model = apps.get_model(model_label)
+            qs = Model.objects.all()
+            if not options['force']:
+                qs = qs.filter(_config_context_data__isnull=True)
+
+            # Annotate so each instance renders from the same aggregated subquery the on-demand path
+            # uses, avoiding N additional queries per object.
+            qs = qs.annotate_config_context_data()
+
+            self.stdout.write(f'Rendering config context for {qs.count()} {model_label} object(s)...')
+
+            rendered = 0
+            for obj in qs.iterator(chunk_size=RENDER_CONFIG_CONTEXT_CHUNK_SIZE):
+                # Capture the generation we render against and write the result back only if no
+                # invalidation has bumped it in the meantime (compare-and-set). This mirrors
+                # RenderConfigContextJob and ensures that running this command on a live system
+                # cannot clobber a concurrent invalidation with a stale render (which would leave a
+                # populated-but-stale cache that the background sweep would then skip).
+                generation = obj._config_context_generation
+                data = obj.render_config_context()
+                rendered += Model.objects.filter(
+                    pk=obj.pk,
+                    _config_context_generation=generation,
+                ).update(_config_context_data=data)
+
+            self.stdout.write(self.style.SUCCESS(f'  Rendered {rendered} {model_label} object(s).'))
+
+        self.stdout.write(self.style.SUCCESS('Finished.'))

netbox/extras/models/configs.py

@@ -1,9 +1,11 @@
+import copy
 import traceback
 
 import jsonschema
 from django.conf import settings
 from django.core.validators import ValidationError
 from django.db import models
+from django.db.models import Q
 from django.urls import reverse
 from django.utils.translation import gettext_lazy as _
 from jinja2.exceptions import TemplateError
@@ -215,12 +217,133 @@ class ConfigContext(SyncedDataMixin, CloningMixin, CustomLinksMixin, OwnerMixin,
         self.data = self.data_file.get_data()
     sync_data.alters_data = True
 
+    def get_affected_objects(self):
+        """
+        Return a (device_qs, vm_qs) tuple of all Devices and VirtualMachines that fall within this
+        ConfigContext's scope. This is the inverse of ConfigContextQuerySet.get_for_object().
+        Used to determine which pre-rendered context caches must be invalidated when this
+        ConfigContext changes.
+        """
+        from dcim.models import Device
+        from virtualization.models import VirtualMachine
+
+        device_q, vm_q = self._get_affected_object_filters()
+        return (
+            Device.objects.filter(device_q),
+            VirtualMachine.objects.filter(vm_q),
+        )
+
+    def _get_affected_object_filters(self):
+        """
+        Build the Q expressions matching Devices and VirtualMachines in this context's scope.
+        Returns (device_q, vm_q). Does NOT consider `is_active` — callers that need that should
+        check it separately. For invalidation purposes, we want the scope set regardless of
+        whether the context is currently active (toggling is_active also requires invalidation).
+        """
+        from extras.models.tags import TaggedItem
+
+        def _nested_scope_q(m2m, object_path):
+            # Match objects whose `object_path` ltree column is a descendant-or-equal of any node
+            # selected in this nested-group m2m (regions, locations, etc.). This is the inverse of
+            # the forward `<object>__path__ancestor_or_equal` match in ConfigContextQuerySet: there
+            # a CC's node must be an ancestor of the object's node; here the object's node must fall
+            # within a CC node's subtree. Returns None if the m2m is empty (no scope restriction).
+            paths = list(m2m.values_list('path', flat=True))
+            if not paths:
+                return None
+            q = Q()
+            for path in paths:
+                q |= Q(**{f'{object_path}__descendant_or_equal': path})
+            return q
+
+        def _direct_pks(m2m):
+            pks = list(m2m.values_list('pk', flat=True))
+            return pks or None
+
+        # Shared filters (applicable to both Device and VirtualMachine)
+        shared = Q()
+
+        region_q = _nested_scope_q(self.regions, 'site__region__path')
+        if region_q is not None:
+            shared &= region_q
+
+        site_group_q = _nested_scope_q(self.site_groups, 'site__group__path')
+        if site_group_q is not None:
+            shared &= site_group_q
+
+        role_q = _nested_scope_q(self.roles, 'role__path')
+        if role_q is not None:
+            shared &= role_q
+
+        platform_q = _nested_scope_q(self.platforms, 'platform__path')
+        if platform_q is not None:
+            shared &= platform_q
+
+        for m2m, path in (
+            (self.sites, 'site'),
+            (self.cluster_types, 'cluster__type'),
+            (self.cluster_groups, 'cluster__group'),
+            (self.clusters, 'cluster'),
+            (self.tenant_groups, 'tenant__group'),
+            (self.tenants, 'tenant'),
+        ):
+            pks = _direct_pks(m2m)
+            if pks is not None:
+                shared &= Q(**{f'{path}__in': pks})
+
+        # Tag-scoped contexts: object must be tagged with at least one of the context's tags
+        tag_pks = _direct_pks(self.tags)
+
+        device_q = Q(shared)
+        vm_q = Q(shared)
+
+        # Device-only filters: location (nested/ltree) and device_type (direct)
+        location_q = _nested_scope_q(self.locations, 'location__path')
+        if location_q is not None:
+            device_q &= location_q
+        device_type_pks = _direct_pks(self.device_types)
+        if device_type_pks is not None:
+            device_q &= Q(device_type__in=device_type_pks)
+        # For VMs, locations and device_types must be empty for the context to apply
+        if location_q is not None or device_type_pks is not None:
+            vm_q &= Q(pk__in=())
+
+        if tag_pks is not None:
+            device_tagged = TaggedItem.objects.filter(
+                tag_id__in=tag_pks,
+                content_type__app_label='dcim',
+                content_type__model='device',
+            ).values_list('object_id', flat=True)
+            vm_tagged = TaggedItem.objects.filter(
+                tag_id__in=tag_pks,
+                content_type__app_label='virtualization',
+                content_type__model='virtualmachine',
+            ).values_list('object_id', flat=True)
+            device_q &= Q(pk__in=device_tagged)
+            vm_q &= Q(pk__in=vm_tagged)
+
+        return device_q, vm_q
+
 
 class ConfigContextModel(models.Model):
     """
     A model which includes local configuration context data. This local data will override any inherited data from
     ConfigContexts.
     """
+    # Pre-rendered config context cache. NULL means "invalidated; render on demand". Populated by
+    # extras.jobs.RenderConfigContextJob in the background.
+    _config_context_data = models.JSONField(
+        blank=True,
+        null=True,
+        editable=False,
+    )
+    # Monotonic counter bumped each time the cache is invalidated. The background renderer captures
+    # this value before rendering and only writes the result back if it is unchanged, so a fresh
+    # invalidation that lands mid-render is never overwritten by a stale value (compare-and-set).
+    _config_context_generation = models.PositiveBigIntegerField(
+        default=0,
+        editable=False,
+    )
     local_context_data = models.JSONField(
         blank=True,
         null=True,
@@ -233,9 +356,25 @@ class ConfigContextModel(models.Model):
         abstract = True
 
     def get_config_context(self):
+        """
+        Return the merged config context for this object. If a pre-rendered cache is present
+        (`_config_context_data`), return a copy of it. Otherwise, fall back to rendering on demand.
+
+        The returned dict is always safe for callers to mutate (e.g. ObjectRenderConfigView merges
+        in additional context with .update()): the cached blob is deep-copied so mutations cannot
+        leak back into this instance's in-memory cache, matching the fresh-dict guarantee of the
+        on-demand render path.
+        """
+        cached = getattr(self, '_config_context_data', None)
+        if cached is not None:
+            return copy.deepcopy(cached)
+        return self.render_config_context()
+
+    def render_config_context(self):
         """
         Compile all config data, overwriting lower-weight values with higher-weight values where a collision occurs.
-        Return the rendered configuration context for a device or VM.
+        Return the rendered configuration context for a device or VM. This bypasses the pre-rendered cache
+        (`_config_context_data`); use get_config_context() for the cached read path.
         """
         data = {}
 
@@ -264,6 +403,15 @@ class ConfigContextModel(models.Model):
                 {'local_context_data': _('JSON data must be in object form. Example:') + ' {"foo": 123}'}
             )
 
+    def serialize_object(self, exclude=None):
+        # Exclude the pre-rendered cache and its generation counter from change-log snapshots;
+        # they are derived fields and would otherwise produce noisy diffs.
+        exclude = list(exclude or [])
+        for field in ('_config_context_data', '_config_context_generation'):
+            if field not in exclude:
+                exclude.append(field)
+        return super().serialize_object(exclude=exclude)
+
 
 #
 # Config templates

netbox/extras/querysets.py

@@ -1,5 +1,5 @@
 from django.contrib.postgres.aggregates import JSONBAgg
-from django.db.models import OuterRef, Q, Subquery
+from django.db.models import Case, JSONField, OuterRef, Q, Subquery, When
 
 from extras.models.tags import TaggedItem
 from utilities.query_functions import EmptyGroupByJSONBAgg
@@ -18,6 +18,10 @@ class ConfigContextQuerySet(RestrictedQuerySet):
         """
         Return all applicable ConfigContexts for a given object. Only active ConfigContexts will be included.
 
+        WARNING: This method's scope-matching logic is mirrored (inverted) by ConfigContext.get_affected_objects(),
+        which powers cache invalidation. Any change to the matching criteria here MUST be applied there as well, or
+        pre-rendered config context caches will go stale. See extras/models/configs.py.
+
         Args:
           aggregate_data: If True, use the JSONBAgg aggregate function to return only the list of JSON data objects
         """
@@ -84,20 +88,39 @@ class ConfigContextModelQuerySet(RestrictedQuerySet):
     This offers a substantial performance gain over ConfigContextQuerySet.get_for_object() when dealing with
     multiple objects. This allows the annotation to be entirely optional.
     """
-    def annotate_config_context_data(self):
+    def annotate_config_context_data(self, only_invalidated=False):
         """
-        Attach the subquery annotation to the base queryset
+        Attach the subquery annotation to the base queryset.
+
+        Args:
+            only_invalidated: If True, evaluate the (expensive) aggregation subquery only for rows
+                whose pre-rendered cache (`_config_context_data`) is NULL, returning NULL for rows
+                that already have a populated cache. This is the list/detail read-path optimization:
+                warm rows are served from the cache by ConfigContextModel.get_config_context() and
+                never consult this annotation, so computing it for them is wasted work. PostgreSQL
+                short-circuits CASE branches, so the correlated SubPlan is not executed for warm
+                rows.
+
+                NOTE: With only_invalidated=True the annotation is NULL for warm rows. It is only
+                safe to read via get_config_context() (which short-circuits on the cache before
+                touching the annotation). Do NOT call render_config_context() directly on a row
+                annotated this way, or a warm row would render an empty context.
         """
         from extras.models import ConfigContext
-        return self.annotate(
-            config_context_data=Subquery(
-                ConfigContext.objects.filter(
-                    self._get_config_context_filters()
-                ).annotate(
-                    _data=EmptyGroupByJSONBAgg('data', order_by=['weight', 'name'])
-                ).values("_data").order_by()
-            )
+        subquery = Subquery(
+            ConfigContext.objects.filter(
+                self._get_config_context_filters()
+            ).annotate(
+                _data=EmptyGroupByJSONBAgg('data', order_by=['weight', 'name'])
+            ).values("_data").order_by()
         )
+        if only_invalidated:
+            subquery = Case(
+                When(_config_context_data__isnull=True, then=subquery),
+                default=None,
+                output_field=JSONField(),
+            )
+        return self.annotate(config_context_data=subquery)
 
     def _get_config_context_filters(self):
         # Construct the set of Q objects for the specific object types

netbox/extras/signals.py

@@ -12,7 +12,13 @@ from netbox.signals import post_clean
 from utilities.data import get_config_value_ci
 from utilities.exceptions import AbortRequest
 
-from .models import CustomField, TaggedItem
+from .cache import (
+    invalidate_config_context_for_configcontext,
+    invalidate_config_context_for_objects,
+    invalidate_for_scope_delta,
+)
+from .constants import CC_FIELDS_BY_MODEL
+from .models import ConfigContext, CustomField, TaggedItem
 from .utils import run_validators
 
 #
@@ -90,6 +96,289 @@ def validate_assigned_tags(sender, instance, action, model, pk_set, **kwargs):
             raise AbortRequest(f"Tag {tag} cannot be assigned to {ct.model} objects.")
 
 
+#
+# Config context cache invalidation
+#
+
+@receiver(post_save, sender=ConfigContext)
+def invalidate_on_configcontext_save(sender, instance, **kwargs):
+    """
+    Whenever a ConfigContext's scalar fields change (e.g. `data`, `weight`, `is_active`),
+    invalidate the caches of all Devices/VMs currently in scope. M2M scope changes are handled
+    separately by invalidate_on_configcontext_m2m_change().
+    """
+    invalidate_config_context_for_configcontext(instance)
+
+
+@receiver(pre_delete, sender=ConfigContext)
+def invalidate_on_configcontext_delete(sender, instance, **kwargs):
+    """
+    Before a ConfigContext is deleted, invalidate the caches of all Devices/VMs currently in
+    scope. The scope is still readable here (pre_delete fires before the row and its M2M rows
+    are removed).
+    """
+    invalidate_config_context_for_configcontext(instance)
+
+
+def invalidate_on_configcontext_m2m_change(sender, instance, action, pk_set, scope_field, **kwargs):
+    """
+    Whenever a ConfigContext's scope M2M changes, invalidate the caches of all Devices/VMs that
+    were or now are in scope.
+
+    Strategy:
+    - For post_add: the current scope is broader than (or equal to) the previous scope. Devices
+      newly in scope are caught by invalidating the current affected set.
+    - For post_remove: the current scope is narrower. We must also invalidate devices that
+      matched only via the just-removed scope items.
+    - For post_clear: the scope is now empty (matches all). The current full affected set is the
+      broadest possible for this attribute; invalidating it suffices.
+    """
+    if action not in ('post_add', 'post_remove', 'post_clear'):
+        return
+
+    # Always invalidate based on the current (post-change) scope.
+    invalidate_config_context_for_configcontext(instance)
+
+    # For post_remove, also invalidate devices/VMs that matched via the removed scope items.
+    if action == 'post_remove' and pk_set:
+        invalidate_for_scope_delta(scope_field, pk_set)
+
+
+def _connect_configcontext_m2m_handlers():
+    """
+    Wire `invalidate_on_configcontext_m2m_change` to every ConfigContext scope M2M's through
+    model. The set of scope M2Ms is introspected from the model so new ones are picked up
+    automatically. The receiver is curried with `scope_field` to identify which attribute changed.
+    """
+    for m2m_field in ConfigContext._meta.many_to_many:
+        field_name = m2m_field.name
+        through = getattr(ConfigContext, field_name).through
+
+        def _handler(sender, instance, action, pk_set, _field=field_name, **kwargs):
+            invalidate_on_configcontext_m2m_change(
+                sender=sender,
+                instance=instance,
+                action=action,
+                pk_set=pk_set,
+                scope_field=_field,
+                **kwargs,
+            )
+
+        m2m_changed.connect(_handler, sender=through, weak=False)
+
+
+_connect_configcontext_m2m_handlers()
+
+
+def _changed_fields(instance, fields):
+    """
+    Return True if any of `fields` differs between the prechange snapshot and the current state.
+    If no snapshot exists (e.g. object loaded fresh from DB and saved without a snapshot), assume
+    we cannot tell what changed and conservatively return True. The cost is one extra background
+    re-render per non-instrumented save; the cost of returning False would be stale caches.
+    """
+    snapshot = getattr(instance, '_prechange_snapshot', None)
+    if not snapshot:
+        return True
+    for field in fields:
+        # Snapshot keys mirror Django's JSON serializer: FK ids are stored under the bare name
+        # (no `_id` suffix). Convert.
+        snap_key = field[:-3] if field.endswith('_id') else field
+        if snapshot.get(snap_key) != getattr(instance, field, None):
+            return True
+    return False
+
+
+def _make_object_save_handler(model_label):
+    fields = CC_FIELDS_BY_MODEL[model_label]
+
+    def _handler(sender, instance, created, **kwargs):
+        # On creation, enqueue a render so the new object's cache is warmed promptly (there is no
+        # recurring sweep). On update, only invalidate when a scope-relevant field actually changed.
+        if created or _changed_fields(instance, fields):
+            invalidate_config_context_for_objects(model_label, [instance.pk])
+
+    return _handler
+
+
+def _connect_object_save_handlers():
+    from django.apps import apps as django_apps
+
+    for model_label in CC_FIELDS_BY_MODEL:
+        Model = django_apps.get_model(model_label)
+        post_save.connect(_make_object_save_handler(model_label), sender=Model, weak=False)
+
+
+_connect_object_save_handlers()
+
+
+@receiver(m2m_changed, sender=TaggedItem)
+def invalidate_on_device_vm_tag_change(sender, instance, action, **kwargs):
+    """
+    When tags are added or removed on a Device/VM, invalidate that object's cache.
+    """
+    if action not in ('post_add', 'post_remove', 'post_clear'):
+        return
+    from dcim.models import Device
+    from virtualization.models import VirtualMachine
+
+    if isinstance(instance, Device):
+        invalidate_config_context_for_objects('dcim.device', [instance.pk])
+    elif isinstance(instance, VirtualMachine):
+        invalidate_config_context_for_objects('virtualization.virtualmachine', [instance.pk])
+
+
+# Upstream object changes that affect ConfigContext matching even when the Device/VM itself is
+# untouched. Two patterns are handled:
+#
+#   1. Direct FK changes (Site.region, Cluster.type, Tenant.group, ...): invalidate the caches of
+#      Devices/VMs that reference the changed object.
+#   2. Ltree reparents (Region.parent, SiteGroup.parent, ...): invalidate every Device/VM whose
+#      attribute resolves into the changed node's subtree, because the ancestor list used by the
+#      matching query has shifted.
+
+
+def _make_direct_upstream_handler(fields, device_lookup, vm_lookup):
+    def _handler(sender, instance, created, **kwargs):
+        if created or not _changed_fields(instance, fields):
+            return
+        from dcim.models import Device
+        from virtualization.models import VirtualMachine
+
+        if device_lookup:
+            invalidate_config_context_for_objects(
+                'dcim.device',
+                Device.objects.filter(**{device_lookup: instance.pk}).values_list('pk', flat=True),
+            )
+        if vm_lookup:
+            invalidate_config_context_for_objects(
+                'virtualization.virtualmachine',
+                VirtualMachine.objects.filter(**{vm_lookup: instance.pk}).values_list('pk', flat=True),
+            )
+
+    return _handler
+
+
+def _make_reparent_handler(device_attr, vm_attr):
+    def _handler(sender, instance, created, **kwargs):
+        if created or not _changed_fields(instance, ('parent_id',)):
+            return
+        from dcim.models import Device
+        from virtualization.models import VirtualMachine
+
+        # The ltree triggers rewrite `path` server-side during the UPDATE, but LtreeModel.save()
+        # only refreshes the in-memory value AFTER post_save fires — so `instance.path` is still
+        # the pre-move value here. Re-read the node's current path from the DB to enumerate its
+        # (post-move) subtree. The set of node PKs is invariant under a move; only their paths
+        # shift, so this matches the same Devices/VMs regardless of timing.
+        model = type(instance)
+        node_path = model.objects.filter(pk=instance.pk).values_list('path', flat=True).first()
+        if node_path is None:
+            return
+        subtree_pks = list(
+            model.objects.filter(path__descendant_or_equal=node_path).values_list('pk', flat=True)
+        )
+
+        if device_attr:
+            invalidate_config_context_for_objects(
+                'dcim.device',
+                Device.objects.filter(**{device_attr: subtree_pks}).values_list('pk', flat=True),
+            )
+        if vm_attr:
+            invalidate_config_context_for_objects(
+                'virtualization.virtualmachine',
+                VirtualMachine.objects.filter(**{vm_attr: subtree_pks}).values_list('pk', flat=True),
+            )
+
+    return _handler
+
+
+def _connect_upstream_handlers():
+    from django.apps import apps as django_apps
+
+    # (app, model, fields_to_watch, device_lookup, vm_lookup)
+    direct_triggers = (
+        ('dcim', 'Site', ('region_id', 'group_id'), 'site_id', 'site_id'),
+        ('dcim', 'Location', ('site_id',), 'location_id', None),
+        ('virtualization', 'Cluster', ('type_id', 'group_id', 'site_id'), 'cluster_id', 'cluster_id'),
+        ('tenancy', 'Tenant', ('group_id',), 'tenant_id', 'tenant_id'),
+    )
+    for app, name, fields, device_lookup, vm_lookup in direct_triggers:
+        Model = django_apps.get_model(app, name)
+        post_save.connect(
+            _make_direct_upstream_handler(fields, device_lookup, vm_lookup),
+            sender=Model,
+            weak=False,
+        )
+
+    # (app, model, device_attr_path__in, vm_attr_path__in)
+    reparent_triggers = (
+        ('dcim', 'Region', 'site__region__in', 'site__region__in'),
+        ('dcim', 'SiteGroup', 'site__group__in', 'site__group__in'),
+        ('dcim', 'DeviceRole', 'role__in', 'role__in'),
+        ('dcim', 'Platform', 'platform__in', 'platform__in'),
+        ('dcim', 'Location', 'location__in', None),
+    )
+    for app, name, device_attr, vm_attr in reparent_triggers:
+        Model = django_apps.get_model(app, name)
+        post_save.connect(
+            _make_reparent_handler(device_attr, vm_attr),
+            sender=Model,
+            weak=False,
+        )
+
+
+_connect_upstream_handlers()
+
+
+# Deletion of an upstream object referenced by a Device/VM via a SET_NULL foreign key (or by a
+# Site/Tenant the object belongs to) silently nulls that FK with a bulk UPDATE that emits no
+# post_save signal, so the object-save handlers above never fire. We therefore invalidate on
+# pre_delete, while the references are still resolvable.
+#
+# Only SET_NULL relationships matter here: PROTECT relationships (Device.role/tenant/site,
+# VM.cluster/site/role/tenant, etc.) cannot be deleted while a Device/VM references them, so no
+# stale cache can result. The SET_NULL feeders into ConfigContext matching are:
+#   - Platform        (Device.platform, VM.platform)
+#   - Cluster         (Device.cluster)            -> also covers cluster_type/cluster_group scopes
+#   - Region          (Site.region)
+#   - SiteGroup       (Site.group)
+#   - TenantGroup     (Tenant.group)
+#
+# We reuse invalidate_for_scope_delta(), which resolves the full set of Devices/VMs reachable via
+# the given scope dimension (descendants included for nested/ltree models), exactly matching the objects
+# whose FK is about to be nulled.
+
+def _make_upstream_delete_handler(scope_field):
+    def _handler(sender, instance, **kwargs):
+        invalidate_for_scope_delta(scope_field, [instance.pk])
+
+    return _handler
+
+
+def _connect_upstream_delete_handlers():
+    from django.apps import apps as django_apps
+
+    # (app, model, scope_field)
+    delete_triggers = (
+        ('dcim', 'Platform', 'platforms'),
+        ('dcim', 'Region', 'regions'),
+        ('dcim', 'SiteGroup', 'site_groups'),
+        ('virtualization', 'Cluster', 'clusters'),
+        ('tenancy', 'TenantGroup', 'tenant_groups'),
+    )
+    for app, name, scope_field in delete_triggers:
+        Model = django_apps.get_model(app, name)
+        pre_delete.connect(
+            _make_upstream_delete_handler(scope_field),
+            sender=Model,
+            weak=False,
+        )
+
+
+_connect_upstream_delete_handlers()
+
+
 #
 # Event rules
 #

netbox/extras/tests/test_configcontext_cache.py

@@ -0,0 +1,709 @@
+from unittest import mock
+
+from django.db import connection
+from django.db.models import F
+from django.test import TestCase
+from django.test.utils import CaptureQueriesContext
+
+from core.models import Job
+from dcim.models import Device, DeviceRole, DeviceType, Location, Manufacturer, Platform, Region, Site, SiteGroup
+from extras.cache import invalidate_config_context_for_objects
+from extras.jobs import RenderConfigContextJob
+from extras.models import ConfigContext, Tag
+from tenancy.models import Tenant, TenantGroup
+from virtualization.models import Cluster, ClusterGroup, ClusterType, VirtualMachine
+
+
+def _set_cache(device, value):
+    """Manually set the cache field and refresh the in-memory instance so subsequent save()
+    calls don't overwrite the DB value with stale in-memory state."""
+    type(device).objects.filter(pk=device.pk).update(_config_context_data=value)
+    device.refresh_from_db()
+
+
+def _get_cache(device):
+    device.refresh_from_db()
+    return device._config_context_data
+
+
+class ConfigContextCacheReadPathTest(TestCase):
+    """
+    get_config_context() must return the cached `_config_context_data` blob when present, and
+    fall back to the on-demand render path when it is NULL.
+    """
+
+    @classmethod
+    def setUpTestData(cls):
+        manufacturer = Manufacturer.objects.create(name='Manufacturer', slug='mfr')
+        cls.devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='DT', slug='dt')
+        cls.role = DeviceRole.objects.create(name='Role', slug='role')
+        cls.site = Site.objects.create(name='Site', slug='site')
+        cls.device = Device.objects.create(
+            name='Device', device_type=cls.devicetype, role=cls.role, site=cls.site
+        )
+
+    def test_cached_value_is_returned(self):
+        cached = {'cached': True, 'value': 42}
+        _set_cache(self.device, cached)
+        device = Device.objects.get(pk=self.device.pk)
+        self.assertEqual(device.get_config_context(), cached)
+
+    def test_null_cache_falls_back_to_render(self):
+        ConfigContext.objects.create(name='CC', weight=100, data={'rendered': True})
+        device = Device.objects.get(pk=self.device.pk)
+        self.assertIsNone(device._config_context_data)
+        self.assertEqual(device.get_config_context(), {'rendered': True})
+
+    def test_render_matches_legacy_path(self):
+        ConfigContext.objects.create(name='A', weight=100, data={'a': 1})
+        ConfigContext.objects.create(name='B', weight=200, data={'a': 2, 'b': 3})
+
+        device = Device.objects.get(pk=self.device.pk)
+        on_demand = device.render_config_context()
+        _set_cache(device, on_demand)
+        self.assertEqual(device.get_config_context(), on_demand)
+
+
+class ConfigContextInvalidationTest(TestCase):
+
+    @classmethod
+    def setUpTestData(cls):
+        manufacturer = Manufacturer.objects.create(name='Mfr', slug='mfr')
+        cls.devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='DT', slug='dt')
+        cls.role = DeviceRole.objects.create(name='Role', slug='role')
+        cls.role2 = DeviceRole.objects.create(name='Role 2', slug='role-2')
+        cls.site1 = Site.objects.create(name='Site 1', slug='site-1')
+        cls.site2 = Site.objects.create(name='Site 2', slug='site-2')
+        cls.device_in_scope = Device.objects.create(
+            name='In scope', device_type=cls.devicetype, role=cls.role, site=cls.site1,
+        )
+        cls.device_out_of_scope = Device.objects.create(
+            name='Out of scope', device_type=cls.devicetype, role=cls.role, site=cls.site2,
+        )
+
+    def setUp(self):
+        # Pre-populate caches for both devices.
+        _set_cache(self.device_in_scope, {'cached': True})
+        _set_cache(self.device_out_of_scope, {'cached': True})
+
+    def test_configcontext_save_invalidates_in_scope_only(self):
+        cc = ConfigContext.objects.create(name='CC', weight=100, data={'x': 1})
+        cc.sites.add(self.site1)
+        # Re-populate caches (the create + m2m add above already triggered invalidations).
+        _set_cache(self.device_in_scope, {'cached': True})
+        _set_cache(self.device_out_of_scope, {'cached': True})
+
+        cc.data = {'x': 2}
+        cc.save()
+
+        self.assertIsNone(_get_cache(self.device_in_scope))
+        self.assertEqual(_get_cache(self.device_out_of_scope), {'cached': True})
+
+    def test_configcontext_delete_invalidates_in_scope(self):
+        cc = ConfigContext.objects.create(name='CC', weight=100, data={'x': 1})
+        cc.sites.add(self.site1)
+        _set_cache(self.device_in_scope, {'cached': True})
+        _set_cache(self.device_out_of_scope, {'cached': True})
+
+        cc.delete()
+
+        self.assertIsNone(_get_cache(self.device_in_scope))
+        self.assertEqual(_get_cache(self.device_out_of_scope), {'cached': True})
+
+    def test_m2m_post_add_invalidates_newly_in_scope(self):
+        cc = ConfigContext.objects.create(name='CC', weight=100, data={'x': 1})
+        _set_cache(self.device_in_scope, {'cached': True})
+
+        cc.sites.add(self.site1)
+
+        self.assertIsNone(_get_cache(self.device_in_scope))
+
+    def test_m2m_post_remove_invalidates_previously_in_scope(self):
+        cc = ConfigContext.objects.create(name='CC', weight=100, data={'x': 1})
+        cc.sites.add(self.site1)
+        _set_cache(self.device_in_scope, {'cached': True})
+
+        cc.sites.remove(self.site1)
+
+        self.assertIsNone(_get_cache(self.device_in_scope))
+
+    def test_device_role_change_invalidates(self):
+        self.device_in_scope.snapshot()
+        self.device_in_scope.role = self.role2
+        self.device_in_scope.save()
+
+        self.assertIsNone(_get_cache(self.device_in_scope))
+
+    def test_device_serial_change_does_not_invalidate(self):
+        # Refresh first so the in-memory instance has the cached value (avoids save() writing
+        # stale NULL back to the DB).
+        self.device_in_scope.refresh_from_db()
+        self.device_in_scope.snapshot()
+        self.device_in_scope.serial = 'ABC123'
+        self.device_in_scope.save()
+
+        self.assertEqual(_get_cache(self.device_in_scope), {'cached': True})
+
+    def test_device_tag_add_invalidates(self):
+        tag = Tag.objects.create(name='Tag', slug='tag')
+        self.device_in_scope.tags.add(tag)
+
+        self.assertIsNone(_get_cache(self.device_in_scope))
+
+
+class ConfigContextUpstreamDeleteInvalidationTest(TestCase):
+    """
+    Deleting an upstream object referenced by a Device/VM via a SET_NULL FK nulls that FK with a
+    bulk UPDATE that emits no post_save signal. The pre_delete handlers must invalidate the
+    affected caches before the references are cleared.
+    """
+
+    @classmethod
+    def setUpTestData(cls):
+        manufacturer = Manufacturer.objects.create(name='Mfr', slug='mfr')
+        cls.devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='DT', slug='dt')
+        cls.role = DeviceRole.objects.create(name='Role', slug='role')
+
+    def test_platform_delete_invalidates(self):
+        platform = Platform.objects.create(name='Platform', slug='platform')
+        site = Site.objects.create(name='Site', slug='site')
+        device = Device.objects.create(
+            name='Device', device_type=self.devicetype, role=self.role, site=site, platform=platform,
+        )
+        _set_cache(device, {'cached': True})
+
+        platform.delete()
+
+        self.assertIsNone(_get_cache(device))
+
+    def test_region_delete_invalidates(self):
+        region = Region.objects.create(name='Region', slug='region')
+        site = Site.objects.create(name='Site', slug='site', region=region)
+        device = Device.objects.create(
+            name='Device', device_type=self.devicetype, role=self.role, site=site,
+        )
+        _set_cache(device, {'cached': True})
+
+        region.delete()
+
+        self.assertIsNone(_get_cache(device))
+
+    def test_site_group_delete_invalidates(self):
+        group = SiteGroup.objects.create(name='Group', slug='group')
+        site = Site.objects.create(name='Site', slug='site', group=group)
+        device = Device.objects.create(
+            name='Device', device_type=self.devicetype, role=self.role, site=site,
+        )
+        _set_cache(device, {'cached': True})
+
+        group.delete()
+
+        self.assertIsNone(_get_cache(device))
+
+
+class RenderConfigContextJobTest(TestCase):
+
+    @classmethod
+    def setUpTestData(cls):
+        manufacturer = Manufacturer.objects.create(name='Mfr', slug='mfr')
+        devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='DT', slug='dt')
+        role = DeviceRole.objects.create(name='Role', slug='role')
+        site = Site.objects.create(name='Site', slug='site')
+        cls.device = Device.objects.create(
+            name='Device', device_type=devicetype, role=role, site=site,
+        )
+        ConfigContext.objects.create(name='CC', weight=100, data={'foo': 'bar'})
+
+    def _make_runner(self):
+        runner = RenderConfigContextJob.__new__(RenderConfigContextJob)
+        runner.job = mock.Mock()
+        runner.logger = mock.Mock()
+        return runner
+
+    def test_job_populates_cache(self):
+        Device.objects.filter(pk=self.device.pk).update(_config_context_data=None)
+        self._make_runner()._render_for_model('dcim.device', pks=[self.device.pk])
+
+        self.device.refresh_from_db()
+        self.assertEqual(self.device._config_context_data, {'foo': 'bar'})
+
+    def test_job_idempotent_on_repopulated_cache(self):
+        runner = self._make_runner()
+        runner._render_for_model('dcim.device', pks=[self.device.pk])
+        self.device.refresh_from_db()
+        first = self.device._config_context_data
+
+        Device.objects.filter(pk=self.device.pk).update(_config_context_data=None)
+        runner._render_for_model('dcim.device', pks=[self.device.pk])
+        self.device.refresh_from_db()
+        self.assertEqual(self.device._config_context_data, first)
+
+    def test_job_skips_stale_write_on_concurrent_invalidation(self):
+        """
+        If an invalidation bumps the generation counter after the renderer captured it but before
+        the result is written, the compare-and-set write must be rejected so a stale value is never
+        persisted (the row stays NULL for the follow-up job to re-render).
+        """
+        Device.objects.filter(pk=self.device.pk).update(
+            _config_context_data=None, _config_context_generation=1
+        )
+        runner = self._make_runner()
+
+        def racing_render(device_self):
+            # Simulate a concurrent invalidation committing mid-render.
+            Device.objects.filter(pk=device_self.pk).update(
+                _config_context_generation=F('_config_context_generation') + 1
+            )
+            return {'stale': True}
+
+        with mock.patch.object(Device, 'render_config_context', autospec=True, side_effect=racing_render):
+            runner._render_for_model('dcim.device', pks=[self.device.pk])
+
+        self.device.refresh_from_db()
+        self.assertIsNone(self.device._config_context_data)
+        self.assertEqual(self.device._config_context_generation, 2)
+
+        # A subsequent render against the current generation succeeds.
+        runner._render_for_model('dcim.device', pks=[self.device.pk])
+        self.device.refresh_from_db()
+        self.assertEqual(self.device._config_context_data, {'foo': 'bar'})
+
+    def test_run_rescans_for_caches_nulled_mid_sweep(self):
+        """
+        An invalidation that commits while this job is already RUNNING coalesces into it (rather
+        than enqueuing a follow-up), so run() must re-scan after each pass. Otherwise a cache that
+        is NULLed after its row has been passed over would be stranded on the on-demand read path
+        with no job left to repopulate it.
+        """
+        device_b = Device.objects.create(
+            name='Device B', device_type=self.device.device_type, role=self.device.role,
+            site=self.device.site,
+        )
+        # Device A needs rendering; Device B starts populated, so the first scan skips it.
+        Device.objects.filter(pk=self.device.pk).update(_config_context_data=None)
+        Device.objects.filter(pk=device_b.pk).update(_config_context_data={'old': True})
+
+        runner = self._make_runner()
+        real_render = Device.render_config_context
+        nulled = []
+
+        def render_then_null_b(device_self):
+            # While rendering Device A (the first pass), simulate a concurrent invalidation NULLing
+            # Device B's cache after B has already been passed over.
+            if device_self.pk == self.device.pk and not nulled:
+                nulled.append(True)
+                Device.objects.filter(pk=device_b.pk).update(_config_context_data=None)
+            return real_render(device_self)
+
+        with mock.patch.object(Device, 'render_config_context', autospec=True, side_effect=render_then_null_b):
+            runner.run(model_label='dcim.device')
+
+        # Both caches must be populated: A on the first pass, B on the re-scan.
+        self.device.refresh_from_db()
+        device_b.refresh_from_db()
+        self.assertEqual(self.device._config_context_data, {'foo': 'bar'})
+        self.assertEqual(device_b._config_context_data, {'foo': 'bar'})
+
+
+class ConfigContextCacheJobEnqueueTest(TestCase):
+    """
+    Invalidation NULLs the cache synchronously and enqueues the render job on transaction commit.
+    """
+
+    @classmethod
+    def setUpTestData(cls):
+        manufacturer = Manufacturer.objects.create(name='Mfr', slug='mfr')
+        devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='DT', slug='dt')
+        role = DeviceRole.objects.create(name='Role', slug='role')
+        site = Site.objects.create(name='Site', slug='site')
+        cls.device = Device.objects.create(
+            name='Device', device_type=devicetype, role=role, site=site,
+        )
+
+    def test_invalidation_enqueues_render_job_on_commit(self):
+        _set_cache(self.device, {'x': 1})
+        self.assertFalse(Job.objects.filter(name=RenderConfigContextJob.name).exists())
+
+        with self.captureOnCommitCallbacks(execute=True):
+            invalidate_config_context_for_objects('dcim.device', [self.device.pk])
+
+        self.assertTrue(Job.objects.filter(name=RenderConfigContextJob.name).exists())
+
+
+class CacheHelperTest(TestCase):
+
+    @classmethod
+    def setUpTestData(cls):
+        manufacturer = Manufacturer.objects.create(name='Mfr', slug='mfr')
+        devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='DT', slug='dt')
+        role = DeviceRole.objects.create(name='Role', slug='role')
+        site = Site.objects.create(name='Site', slug='site')
+        cls.device = Device.objects.create(
+            name='Device', device_type=devicetype, role=role, site=site,
+        )
+
+    def test_invalidate_for_objects_nulls_cache(self):
+        _set_cache(self.device, {'x': 1})
+        invalidate_config_context_for_objects('dcim.device', [self.device.pk])
+        self.assertIsNone(_get_cache(self.device))
+
+    def test_invalidate_for_objects_bumps_generation(self):
+        _set_cache(self.device, {'x': 1})
+        self.device.refresh_from_db()
+        before = self.device._config_context_generation
+        invalidate_config_context_for_objects('dcim.device', [self.device.pk])
+        self.device.refresh_from_db()
+        self.assertEqual(self.device._config_context_generation, before + 1)
+
+    def test_invalidate_with_empty_args_is_noop(self):
+        invalidate_config_context_for_objects('dcim.device', [])
+
+
+class VirtualMachineInvalidationTest(TestCase):
+    """
+    The invalidation signals must cover VirtualMachine, not just Device.
+    """
+
+    @classmethod
+    def setUpTestData(cls):
+        cls.site1 = Site.objects.create(name='Site 1', slug='site-1')
+        cls.site2 = Site.objects.create(name='Site 2', slug='site-2')
+        cls.role = DeviceRole.objects.create(name='Role', slug='role')
+        cls.role2 = DeviceRole.objects.create(name='Role 2', slug='role-2')
+        clustertype = ClusterType.objects.create(name='CT', slug='ct')
+        cls.cluster = Cluster.objects.create(name='Cluster', type=clustertype)
+        cls.vm_in_scope = VirtualMachine.objects.create(name='In scope', site=cls.site1, role=cls.role)
+        cls.vm_out_of_scope = VirtualMachine.objects.create(name='Out of scope', site=cls.site2, role=cls.role)
+
+    def setUp(self):
+        _set_cache(self.vm_in_scope, {'cached': True})
+        _set_cache(self.vm_out_of_scope, {'cached': True})
+
+    def test_configcontext_save_invalidates_in_scope_vm_only(self):
+        cc = ConfigContext.objects.create(name='CC', weight=100, data={'x': 1})
+        cc.sites.add(self.site1)
+        _set_cache(self.vm_in_scope, {'cached': True})
+        _set_cache(self.vm_out_of_scope, {'cached': True})
+
+        cc.data = {'x': 2}
+        cc.save()
+
+        self.assertIsNone(_get_cache(self.vm_in_scope))
+        self.assertEqual(_get_cache(self.vm_out_of_scope), {'cached': True})
+
+    def test_vm_role_change_invalidates(self):
+        self.vm_in_scope.refresh_from_db()
+        self.vm_in_scope.snapshot()
+        self.vm_in_scope.role = self.role2
+        self.vm_in_scope.save()
+        self.assertIsNone(_get_cache(self.vm_in_scope))
+
+    def test_vm_tag_add_invalidates(self):
+        tag = Tag.objects.create(name='Tag', slug='tag')
+        self.vm_in_scope.tags.add(tag)
+        self.assertIsNone(_get_cache(self.vm_in_scope))
+
+    def test_device_only_scope_does_not_invalidate_vm(self):
+        # A context scoped by device_type can never apply to a VM, so its changes must not touch
+        # VM caches.
+        manufacturer = Manufacturer.objects.create(name='Mfr', slug='mfr')
+        devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='DT', slug='dt')
+        cc = ConfigContext.objects.create(name='CC', weight=100, data={'x': 1})
+        cc.device_types.add(devicetype)
+        _set_cache(self.vm_in_scope, {'cached': True})
+
+        cc.data = {'x': 2}
+        cc.save()
+
+        self.assertEqual(_get_cache(self.vm_in_scope), {'cached': True})
+
+
+class ConfigContextUpstreamChangeInvalidationTest(TestCase):
+    """
+    Changes to intermediate/hierarchical objects (not the Device/VM itself) must invalidate the
+    affected caches: direct FK changes on an intermediate model, and ltree reparents.
+    """
+
+    @classmethod
+    def setUpTestData(cls):
+        manufacturer = Manufacturer.objects.create(name='Mfr', slug='mfr')
+        cls.devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='DT', slug='dt')
+        cls.role = DeviceRole.objects.create(name='Role', slug='role')
+
+    def _make_device(self, site):
+        device = Device.objects.create(
+            name=f'Device {site.slug}', device_type=self.devicetype, role=self.role, site=site,
+        )
+        _set_cache(device, {'cached': True})
+        return device
+
+    def test_direct_upstream_site_region_change_invalidates(self):
+        region1 = Region.objects.create(name='R1', slug='r1')
+        region2 = Region.objects.create(name='R2', slug='r2')
+        site = Site.objects.create(name='Site', slug='site', region=region1)
+        device = self._make_device(site)
+
+        site.snapshot()
+        site.region = region2
+        site.save()
+
+        self.assertIsNone(_get_cache(device))
+
+    def test_region_reparent_invalidates(self):
+        region_a = Region.objects.create(name='A', slug='a')
+        region_b = Region.objects.create(name='B', slug='b')
+        site = Site.objects.create(name='Site', slug='site', region=region_b)
+        device = self._make_device(site)
+
+        region_b.snapshot()
+        region_b.parent = region_a
+        region_b.save()
+
+        self.assertIsNone(_get_cache(device))
+
+    def test_direct_upstream_cluster_type_change_invalidates_vm(self):
+        ct1 = ClusterType.objects.create(name='CT1', slug='ct1')
+        ct2 = ClusterType.objects.create(name='CT2', slug='ct2')
+        cluster = Cluster.objects.create(name='Cluster', type=ct1)
+        site = Site.objects.create(name='Site', slug='site')
+        vm = VirtualMachine.objects.create(name='VM', site=site, role=self.role, cluster=cluster)
+        _set_cache(vm, {'cached': True})
+
+        cluster.snapshot()
+        cluster.type = ct2
+        cluster.save()
+
+        self.assertIsNone(_get_cache(vm))
+
+    def test_direct_upstream_cluster_group_change_invalidates_vm(self):
+        ct = ClusterType.objects.create(name='CT', slug='ct')
+        cg1 = ClusterGroup.objects.create(name='CG1', slug='cg1')
+        cg2 = ClusterGroup.objects.create(name='CG2', slug='cg2')
+        cluster = Cluster.objects.create(name='Cluster', type=ct, group=cg1)
+        site = Site.objects.create(name='Site', slug='site')
+        vm = VirtualMachine.objects.create(name='VM', site=site, role=self.role, cluster=cluster)
+        _set_cache(vm, {'cached': True})
+
+        cluster.snapshot()
+        cluster.group = cg2
+        cluster.save()
+
+        self.assertIsNone(_get_cache(vm))
+
+    def test_direct_upstream_tenant_group_change_invalidates_device(self):
+        tg1 = TenantGroup.objects.create(name='TG1', slug='tg1')
+        tg2 = TenantGroup.objects.create(name='TG2', slug='tg2')
+        tenant = Tenant.objects.create(name='Tenant', slug='tenant', group=tg1)
+        site = Site.objects.create(name='Site', slug='site')
+        device = Device.objects.create(
+            name='Device', device_type=self.devicetype, role=self.role, site=site, tenant=tenant,
+        )
+        _set_cache(device, {'cached': True})
+
+        tenant.snapshot()
+        tenant.group = tg2
+        tenant.save()
+
+        self.assertIsNone(_get_cache(device))
+
+    def test_direct_upstream_tenant_group_change_invalidates_vm(self):
+        tg1 = TenantGroup.objects.create(name='TG1', slug='tg1')
+        tg2 = TenantGroup.objects.create(name='TG2', slug='tg2')
+        tenant = Tenant.objects.create(name='Tenant', slug='tenant', group=tg1)
+        site = Site.objects.create(name='Site', slug='site')
+        vm = VirtualMachine.objects.create(name='VM', site=site, role=self.role, tenant=tenant)
+        _set_cache(vm, {'cached': True})
+
+        tenant.snapshot()
+        tenant.group = tg2
+        tenant.save()
+
+        self.assertIsNone(_get_cache(vm))
+
+    def test_m2m_post_clear_invalidates(self):
+        site = Site.objects.create(name='Site', slug='site')
+        device = self._make_device(site)
+        cc = ConfigContext.objects.create(name='CC', weight=100, data={'x': 1})
+        cc.sites.add(site)
+        _set_cache(device, {'cached': True})
+
+        cc.sites.clear()
+
+        self.assertIsNone(_get_cache(device))
+
+
+class ConfigContextScopeParityTest(TestCase):
+    """
+    The inverse matcher (ConfigContext.get_affected_objects()) must agree exactly with the forward
+    matcher (ConfigContextQuerySet.get_for_object()) across every scope dimension, including
+    hierarchy expansion and device-only dimensions.
+    """
+
+    @classmethod
+    def setUpTestData(cls):
+        # Hierarchies
+        cls.region_parent = Region.objects.create(name='Region Parent', slug='region-parent')
+        cls.region_child = Region.objects.create(name='Region Child', slug='region-child', parent=cls.region_parent)
+        cls.sg_parent = SiteGroup.objects.create(name='SG Parent', slug='sg-parent')
+        cls.sg_child = SiteGroup.objects.create(name='SG Child', slug='sg-child', parent=cls.sg_parent)
+        cls.role_parent = DeviceRole.objects.create(name='Role Parent', slug='role-parent')
+        cls.role_child = DeviceRole.objects.create(name='Role Child', slug='role-child', parent=cls.role_parent)
+        cls.plat_parent = Platform.objects.create(name='Plat Parent', slug='plat-parent')
+        cls.plat_child = Platform.objects.create(name='Plat Child', slug='plat-child', parent=cls.plat_parent)
+        cls.tg = TenantGroup.objects.create(name='TG', slug='tg')
+        cls.tenant = Tenant.objects.create(name='Tenant', slug='tenant', group=cls.tg)
+
+        cls.site_a = Site.objects.create(
+            name='Site A', slug='site-a', region=cls.region_child, group=cls.sg_child
+        )
+        cls.site_b = Site.objects.create(name='Site B', slug='site-b')
+        cls.loc_parent = Location.objects.create(name='Loc Parent', slug='loc-parent', site=cls.site_a)
+        cls.loc_child = Location.objects.create(
+            name='Loc Child', slug='loc-child', site=cls.site_a, parent=cls.loc_parent
+        )
+
+        manufacturer = Manufacturer.objects.create(name='Mfr', slug='mfr')
+        cls.devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='DT', slug='dt')
+        clustertype = ClusterType.objects.create(name='CT', slug='ct')
+        clustergroup = ClusterGroup.objects.create(name='CG', slug='cg')
+        cls.cluster = Cluster.objects.create(name='Cluster', type=clustertype, group=clustergroup)
+        cls.tag = Tag.objects.create(name='Tag', slug='tag')
+
+        cls.d_full = Device.objects.create(
+            name='d-full', device_type=cls.devicetype, role=cls.role_child, site=cls.site_a,
+            location=cls.loc_child, platform=cls.plat_child, tenant=cls.tenant, cluster=cls.cluster,
+        )
+        cls.d_full.tags.add(cls.tag)
+        cls.d_min = Device.objects.create(
+            name='d-min', device_type=cls.devicetype, role=cls.role_parent, site=cls.site_b,
+        )
+        cls.vm_full = VirtualMachine.objects.create(
+            name='vm-full', site=cls.site_a, role=cls.role_child, platform=cls.plat_child,
+            tenant=cls.tenant, cluster=cls.cluster,
+        )
+        cls.vm_full.tags.add(cls.tag)
+        cls.vm_min = VirtualMachine.objects.create(name='vm-min', site=cls.site_b, role=cls.role_parent)
+
+    def _assert_parity(self, cc):
+        device_qs, vm_qs = cc.get_affected_objects()
+        inverse_devices = set(device_qs.values_list('pk', flat=True))
+        inverse_vms = set(vm_qs.values_list('pk', flat=True))
+
+        forward_devices = {
+            d.pk for d in Device.objects.all()
+            if ConfigContext.objects.get_for_object(d).filter(pk=cc.pk).exists()
+        }
+        forward_vms = {
+            v.pk for v in VirtualMachine.objects.all()
+            if ConfigContext.objects.get_for_object(v).filter(pk=cc.pk).exists()
+        }
+        self.assertEqual(inverse_devices, forward_devices, f"device mismatch for {cc.name}")
+        self.assertEqual(inverse_vms, forward_vms, f"VM mismatch for {cc.name}")
+
+    def test_scope_parity_across_dimensions(self):
+        scopes = {
+            'regions': [self.region_parent],         # hierarchy: matches descendant region_child
+            'site_groups': [self.sg_parent],         # hierarchy
+            'sites': [self.site_a],
+            'locations': [self.loc_parent],          # device-only, hierarchy
+            'device_types': [self.devicetype],       # device-only
+            'roles': [self.role_parent],             # hierarchy
+            'platforms': [self.plat_parent],         # hierarchy
+            'cluster_types': [self.cluster.type],
+            'cluster_groups': [self.cluster.group],
+            'clusters': [self.cluster],
+            'tenant_groups': [self.tg],              # direct (immediate group only)
+            'tenants': [self.tenant],
+            'tags': [self.tag],
+        }
+        for i, (field, items) in enumerate(scopes.items()):
+            cc = ConfigContext.objects.create(name=f'cc-{field}', weight=100 + i, data={field: True})
+            getattr(cc, field).set(items)
+            self._assert_parity(cc)
+
+        # A context with no scope matches every object.
+        cc_all = ConfigContext.objects.create(name='cc-all', weight=999, data={'all': True})
+        self._assert_parity(cc_all)
+
+
+class ConfigContextCacheQueryCountTest(TestCase):
+    """
+    Proves the optimization: when caches are warm, get_config_context() issues no per-object
+    queries, whereas the cold fallback path scales with the number of objects.
+    """
+
+    @classmethod
+    def setUpTestData(cls):
+        manufacturer = Manufacturer.objects.create(name='Mfr', slug='mfr')
+        devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='DT', slug='dt')
+        role = DeviceRole.objects.create(name='Role', slug='role')
+        site = Site.objects.create(name='Site', slug='site')
+        for i in range(4):
+            Device.objects.create(name=f'Device {i}', device_type=devicetype, role=role, site=site)
+        ConfigContext.objects.create(name='A', weight=100, data={'a': 1})
+        ConfigContext.objects.create(name='B', weight=200, data={'b': 2})
+
+    def test_warm_cache_avoids_per_object_queries(self):
+        # Cold: caches are NULL, so each object renders on demand.
+        Device.objects.update(_config_context_data=None)
+        cold_devices = list(Device.objects.all())
+        with CaptureQueriesContext(connection) as cold:
+            for device in cold_devices:
+                device.get_config_context()
+
+        # Warm the caches.
+        for device in Device.objects.all():
+            Device.objects.filter(pk=device.pk).update(_config_context_data=device.render_config_context())
+
+        warm_devices = list(Device.objects.all())
+        with CaptureQueriesContext(connection) as warm:
+            for device in warm_devices:
+                device.get_config_context()
+
+        self.assertEqual(len(warm.captured_queries), 0)
+        self.assertGreaterEqual(len(cold.captured_queries), len(cold_devices))
+
+
+class ConditionalConfigContextAnnotationTest(TestCase):
+    """
+    annotate_config_context_data(only_invalidated=True) is the list/detail read-path optimization:
+    it must compute the aggregation only for rows whose cache is NULL and leave it NULL for warm
+    rows, while a single query still serves a mix of warm and cold objects correctly via
+    get_config_context().
+    """
+
+    @classmethod
+    def setUpTestData(cls):
+        manufacturer = Manufacturer.objects.create(name='Mfr', slug='mfr')
+        devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='DT', slug='dt')
+        role = DeviceRole.objects.create(name='Role', slug='role')
+        site = Site.objects.create(name='Site', slug='site')
+        cls.warm = Device.objects.create(name='Warm', device_type=devicetype, role=role, site=site)
+        cls.cold = Device.objects.create(name='Cold', device_type=devicetype, role=role, site=site)
+        ConfigContext.objects.create(name='A', weight=100, data={'a': 1})
+        ConfigContext.objects.create(name='B', weight=200, data={'b': 2})
+
+    def setUp(self):
+        # Warm one device with a sentinel cache, leave the other invalidated.
+        _set_cache(self.warm, {'sentinel': True})
+        Device.objects.filter(pk=self.cold.pk).update(_config_context_data=None)
+
+    def test_annotation_skips_warm_rows(self):
+        rows = {
+            d.pk: d
+            for d in Device.objects.annotate_config_context_data(only_invalidated=True)
+        }
+        # Warm row: the aggregation must not have run; the annotated value is NULL.
+        self.assertIsNone(rows[self.warm.pk].config_context_data)
+        # Cold row: the aggregation ran and produced the ordered list of context data.
+        self.assertEqual(rows[self.cold.pk].config_context_data, [{'a': 1}, {'b': 2}])
+
+    def test_single_query_serves_mixed_warm_and_cold(self):
+        qs = Device.objects.annotate_config_context_data(only_invalidated=True)
+        with CaptureQueriesContext(connection) as ctx:
+            rows = {d.pk: d.get_config_context() for d in qs}
+        # The warm row returns its cached sentinel; the cold row is rendered from the annotation.
+        self.assertEqual(rows[self.warm.pk], {'sentinel': True})
+        self.assertEqual(rows[self.cold.pk], {'a': 1, 'b': 2})
+        # A single query backs the whole page — no per-object fallback to get_for_object().
+        self.assertEqual(len(ctx.captured_queries), 1)

netbox/extras/tests/test_management_commands.py

@@ -5,14 +5,17 @@ from unittest.mock import MagicMock, patch
 from django.contrib.contenttypes.models import ContentType
 from django.core.management import call_command
 from django.core.management.base import CommandError
+from django.db.models import F
 from django.test import TestCase
 
 from dcim.choices import InterfaceTypeChoices
 from dcim.models import Device, DeviceRole, DeviceType, Interface, Manufacturer, Site
 from extras.management.commands import renaturalize, webhook_receiver
 from extras.management.commands.webhook_receiver import WebhookHandler
+from extras.models import ConfigContext
 from users.models import User
 from utilities.fields import NaturalOrderingField
+from virtualization.models import VirtualMachine
 
 
 class ReindexTestCase(TestCase):
@@ -412,6 +415,87 @@ class RunScriptTestCase(TestCase):
         self.assertEqual(enqueue.call_args.kwargs['user'], self.user)
 
 
+class RebuildConfigContextCacheCommandTest(TestCase):
+
+    @classmethod
+    def setUpTestData(cls):
+        manufacturer = Manufacturer.objects.create(name='Mfr', slug='mfr')
+        devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='DT', slug='dt')
+        role = DeviceRole.objects.create(name='Role', slug='role')
+        site = Site.objects.create(name='Site', slug='site')
+        cls.device = Device.objects.create(name='Device', device_type=devicetype, role=role, site=site)
+        cls.vm = VirtualMachine.objects.create(name='VM', site=site, role=role)
+        ConfigContext.objects.create(name='CC', weight=100, data={'foo': 'bar'})
+
+    def test_command_populates_null_caches(self):
+        Device.objects.update(_config_context_data=None)
+        VirtualMachine.objects.update(_config_context_data=None)
+
+        call_command('rebuild_config_context_cache')
+
+        self.device.refresh_from_db()
+        self.vm.refresh_from_db()
+        self.assertEqual(self.device._config_context_data, {'foo': 'bar'})
+        self.assertEqual(self.vm._config_context_data, {'foo': 'bar'})
+
+    def test_command_skips_populated_caches_by_default(self):
+        # Seed a stale value; without --force the command must leave already-populated caches alone.
+        Device.objects.update(_config_context_data={'stale': True})
+        VirtualMachine.objects.update(_config_context_data={'stale': True})
+
+        call_command('rebuild_config_context_cache')
+
+        self.device.refresh_from_db()
+        self.vm.refresh_from_db()
+        self.assertEqual(self.device._config_context_data, {'stale': True})
+        self.assertEqual(self.vm._config_context_data, {'stale': True})
+
+    def test_command_force_rerenders_populated_caches(self):
+        # A stale cache must be re-rendered from scratch when --force is given.
+        Device.objects.update(_config_context_data={'stale': True})
+        VirtualMachine.objects.update(_config_context_data={'stale': True})
+
+        call_command('rebuild_config_context_cache', '--force')
+
+        self.device.refresh_from_db()
+        self.vm.refresh_from_db()
+        self.assertEqual(self.device._config_context_data, {'foo': 'bar'})
+        self.assertEqual(self.vm._config_context_data, {'foo': 'bar'})
+
+    def test_command_reports_rendered_counts(self):
+        Device.objects.update(_config_context_data=None)
+        VirtualMachine.objects.update(_config_context_data=None)
+
+        out = StringIO()
+        call_command('rebuild_config_context_cache', stdout=out)
+        output = out.getvalue()
+
+        self.assertIn('Rendered 1 dcim.device object(s).', output)
+        self.assertIn('Rendered 1 virtualization.virtualmachine object(s).', output)
+        self.assertIn('Finished.', output)
+
+    def test_command_respects_generation_guard(self):
+        """
+        Running the command on a live system must not clobber a concurrent invalidation: if the
+        generation counter is bumped while an object is being rendered, the compare-and-set write
+        is rejected and the cache stays NULL for the background sweep to repopulate.
+        """
+        Device.objects.update(_config_context_data=None)
+
+        def racing_render(device_self):
+            # Simulate a concurrent invalidation committing mid-render.
+            Device.objects.filter(pk=device_self.pk).update(
+                _config_context_generation=F('_config_context_generation') + 1
+            )
+            return {'stale': True}
+
+        with patch.object(Device, 'render_config_context', autospec=True, side_effect=racing_render):
+            call_command('rebuild_config_context_cache')
+
+        self.device.refresh_from_db()
+        self.assertIsNone(self.device._config_context_data)
+
+
 class WebhookReceiverTestCase(TestCase):
     def test_starts_http_server(self):
         out = StringIO()

netbox/virtualization/api/serializers_/virtualmachines.py

@@ -27,7 +27,6 @@ __all__ = (
     'VirtualDiskSerializer',
     'VirtualMachineSerializer',
     'VirtualMachineTypeSerializer',
-    'VirtualMachineWithConfigContextSerializer',
 )
 
 
@@ -76,6 +75,7 @@ class VirtualMachineSerializer(PrimaryModelSerializer):
         fields=[*IPAddressSerializer.Meta.brief_fields, 'nat_inside', 'nat_outside'],
     )
     config_template = ConfigTemplateSerializer(nested=True, required=False, allow_null=True, default=None)
+    config_context = serializers.SerializerMethodField(read_only=True)
 
     # Counter fields
     interface_count = serializers.IntegerField(read_only=True)
@@ -83,19 +83,6 @@ class VirtualMachineSerializer(PrimaryModelSerializer):
 
     class Meta:
         model = VirtualMachine
-        fields = [
-            'id', 'url', 'display_url', 'display', 'name', 'virtual_machine_type', 'role', 'status', 'start_on_boot',
-            'site', 'cluster', 'device', 'platform', 'primary_ip', 'primary_ip4', 'primary_ip6', 'vcpus', 'memory',
-            'disk', 'description', 'serial', 'tenant', 'owner', 'comments', 'tags', 'local_context_data',
-            'config_template', 'custom_fields', 'created', 'last_updated', 'interface_count', 'virtual_disk_count',
-        ]
-        brief_fields = ('id', 'url', 'display', 'name', 'description')
-
-
-class VirtualMachineWithConfigContextSerializer(VirtualMachineSerializer):
-    config_context = serializers.SerializerMethodField()
-
-    class Meta(VirtualMachineSerializer.Meta):
         fields = [
             'id', 'url', 'display_url', 'display', 'name', 'virtual_machine_type', 'role', 'status', 'start_on_boot',
             'site', 'cluster', 'device', 'platform', 'primary_ip', 'primary_ip4', 'primary_ip6', 'vcpus', 'memory',
@@ -103,6 +90,7 @@ class VirtualMachineWithConfigContextSerializer(VirtualMachineSerializer):
             'config_template', 'custom_fields', 'created', 'last_updated', 'interface_count', 'virtual_disk_count',
             'config_context',
         ]
+        brief_fields = ('id', 'url', 'display', 'name', 'description')
 
     @extend_schema_field(serializers.JSONField(allow_null=True))
     def get_config_context(self, obj):

netbox/virtualization/api/views.py

@@ -64,24 +64,9 @@ class VirtualMachineTypeViewSet(NetBoxModelViewSet):
 
 class VirtualMachineViewSet(ConfigContextQuerySetMixin, RenderConfigMixin, NetBoxModelViewSet):
     queryset = VirtualMachine.objects.all()
+    serializer_class = serializers.VirtualMachineSerializer
     filterset_class = filtersets.VirtualMachineFilterSet
 
-    def get_serializer_class(self):
-        """
-        Select the specific serializer based on the request context.
-
-        If the `brief` query param equates to True, return the NestedVirtualMachineSerializer
-
-        If the `exclude` query param includes `config_context` as a value, return the VirtualMachineSerializer
-
-        Else, return the VirtualMachineWithConfigContextSerializer
-        """
-        request = self.get_serializer_context()['request']
-        if self.brief or 'config_context' in request.query_params.get('exclude', []):
-            return serializers.VirtualMachineSerializer
-
-        return serializers.VirtualMachineWithConfigContextSerializer
-
 
 class VMInterfaceViewSet(NetBoxModelViewSet):
     queryset = VMInterface.objects.prefetch_related(

netbox/virtualization/migrations/0057_virtualmachine__config_context_data.py

@@ -0,0 +1,28 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ('virtualization', '0056_virtualmachine_render_config_permission'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='virtualmachine',
+            name='_config_context_data',
+            field=models.JSONField(blank=True, editable=False, null=True),
+        ),
+        migrations.AddField(
+            model_name='virtualmachine',
+            name='_config_context_generation',
+            field=models.PositiveBigIntegerField(default=0, editable=False),
+        ),
+        migrations.AddIndex(
+            model_name='virtualmachine',
+            index=models.Index(
+                condition=models.Q(('_config_context_data__isnull', True)),
+                fields=['id'],
+                name='virtualization_vm_cc_null',
+            ),
+        ),
+    ]

netbox/virtualization/models/virtualmachines.py

@@ -250,6 +250,15 @@ class VirtualMachine(
         ordering = ('name', 'pk')  # Name may be non-unique
         indexes = (
             models.Index(fields=('name', 'id')),  # Default ordering
+            # Partial index supporting the background renderer's scan for objects whose config
+            # context cache has been invalidated (see extras.jobs.RenderConfigContextJob and
+            # extras.cache). Indexing only the NULL-cache rows keeps that lookup cheap on large
+            # tables where the steady state is for nearly every row to be populated.
+            models.Index(
+                fields=('id',),
+                condition=Q(_config_context_data__isnull=True),
+                name='virtualization_vm_cc_null',
+            ),
         )
         constraints = (
             models.UniqueConstraint(

netbox/virtualization/tests/test_api.py

@@ -418,16 +418,6 @@ class VirtualMachineTestCase(APIViewTestCases.APIViewTestCase):
         response = self.client.get(url, **self.header)
         self.assertEqual(response.data['results'][0].get('config_context', {}).get('A'), 1)
 
-    def test_config_context_excluded(self):
-        """
-        Check that config context data can be excluded by passing ?exclude=config_context.
-        """
-        url = reverse('virtualization-api:virtualmachine-list') + '?exclude=config_context'
-        self.add_permissions('virtualization.view_virtualmachine')
-
-        response = self.client.get(url, **self.header)
-        self.assertFalse('config_context' in response.data['results'][0])
-
     def test_unique_name_per_cluster_constraint(self):
         """
         Check that creating a virtual machine with a duplicate name fails.
@@ -544,7 +534,7 @@ class VirtualMachineTestCase(APIViewTestCases.APIViewTestCase):
 
         self.add_permissions('virtualization.view_virtualmachine', 'ipam.view_ipaddress')
         response = self.client.get(
-            f'{self._get_detail_url(virtualmachine)}?exclude=config_context',
+            self._get_detail_url(virtualmachine),
             **self.header,
         )
         self.assertHttpStatus(response, status.HTTP_200_OK)

netbox/virtualization/views.py

@@ -563,7 +563,7 @@ class VirtualMachineVirtualDisksView(generic.ObjectChildrenView):
 
 @register_model_view(VirtualMachine, 'configcontext', path='config-context')
 class VirtualMachineConfigContextView(ObjectConfigContextView):
-    queryset = VirtualMachine.objects.annotate_config_context_data()
+    queryset = VirtualMachine.objects.all()
     base_template = 'virtualization/virtualmachine.html'
     tab = ViewTab(
         label=_('Config Context'),

upgrade.sh

@@ -104,6 +104,11 @@ COMMAND="python3 netbox/manage.py trace_paths --no-input"
 echo "Checking for missing cable paths ($COMMAND)..."
 eval $COMMAND || exit 1
 
+# Pre-render and cache config context data for devices and virtual machines
+COMMAND="python3 netbox/manage.py rebuild_config_context_cache"
+echo "Caching config context data ($COMMAND)..."
+eval $COMMAND || exit 1
+
 # Build the local documentation
 COMMAND="zensical build"
 echo "Building documentation ($COMMAND)..."