Fixes #23010: Defer bulk changes to object data when adding/removing a custom field (#23011)

docs/configuration/system.md

@@ -20,6 +20,8 @@ The maximum number of rows to affect in a single SQL `UPDATE` statement when Net
 
 Must be a positive integer, or `None` to disable chunking and issue each bulk update as a single unbounded statement.
 
+This parameter also determines when a custom field operation is deferred to a background job: creating a field with a default value, or deleting a field, is performed within the request only where the field's assigned object types hold no more than this many objects in total (see [field status](../customization/custom-fields.md#field-status)). Setting it to `None` therefore defers every such operation which affects any object.
+
 ```python
 BULK_UPDATE_CHUNK_SIZE = 5000
 ```

docs/customization/custom-fields.md

@@ -37,7 +37,34 @@ Unless the field has been assigned a default value, creating a custom field does
 
 This matters only if you query the underlying `custom_field_data` JSON directly, for example in a custom script. The field's key is absent from an object's data until a value is assigned to it, so read it with `obj.cf['field_name']` or `obj.custom_field_data.get('field_name')` rather than by direct subscript.
 
-Assigning a default value, by contrast, does write that value to every existing object at the time the field is created, so that objects can be filtered by it immediately. On a model with a very large number of objects, this can take some time. Note that a default added to a field which already exists is _not_ backfilled: objects with no value continue to report none until they are next saved.
+Assigning a default value, by contrast, does write that value to every existing object at the time the field is created, so that objects can be filtered by it immediately. Note that a default added to a field which already exists is _not_ backfilled: objects with no value continue to report none until they are next saved.
+
+### Field Status
+
+!!! info "This behavior was introduced in NetBox v4.7.0."
+
+Creating a custom field with a default value, and deleting a custom field, both require rewriting the stored data of the objects the field applies to. Where the field is assigned to a large number of objects, this cannot be completed within the request, so it is handed to a background job instead and the field reports its status accordingly:
+
+| Status | Meaning |
+| ------ | ------- |
+| Active | The field is live and available for use. |
+| Provisioning | The field's default value is being written to existing objects. |
+| Deleting | The field's data is being removed from existing objects. |
+
+Whether a background job is required is determined by the total number of objects of the field's assigned object types, measured against the [`BULK_UPDATE_CHUNK_SIZE`](../configuration/system.md#bulk_update_chunk_size) configuration parameter — not by how many of those objects actually hold a value for the field. Deleting a field assigned to a large table is therefore deferred even where the field holds no data at all: NetBox cannot count the objects holding a value without scanning the entire table, which is the cost the threshold exists to avoid.
+
+A field is live only while active. During provisioning or deletion it does not appear on objects, in forms, in filters, or in either API, and its stored data is read and written by nothing but the job responsible for it; it becomes available (or disappears entirely) once the job completes. Objects created in the meantime are unaffected — a field being provisioned still supplies its default to new objects.
+
+A field which is not active cannot be modified while its job runs, as its configuration must not change under the job rewriting its data. This includes assigning it further object types, and unassigning those it already carries: such a change is rejected until the field is live again.
+
+A field pending deletion continues to occupy its name until its data has been removed, so that a new field cannot be created — and an existing field cannot be renamed — to a name whose old values are still present on objects.
+
+These operations require a running [background worker](../features/background-jobs.md) (`rqworker`). A field left mid-operation, for example because no worker was running or because its job failed, remains in its pending status until that job runs to completion.
+
+Such a field can always be deleted, whichever status it holds. Deleting one already pending deletion queues a fresh job to finish removing its data. A field left provisioning has no equivalent in-application retry: requeue its job from the background queues (**Admin > System > Background Tasks**, which requires a staff account), or delete the field and create it again.
+
+!!! note
+    Unassigning an object type from a custom field still removes the field's data from those objects immediately, and remains subject to the request timeout on very large tables. The same applies to renaming a custom field.
 
 ### Filtering
 

docs/models/extras/customfield.md

@@ -12,6 +12,10 @@ Select the NetBox object type or types to which this custom field applies.
 
 The raw field name. This will be used in the database and API, and should consist only of alphanumeric characters and underscores. (Use the `label` field to designate a human-friendly name for the custom field.)
 
+### Status
+
+The field's lifecycle state: `active`, `provisioning`, or `deleting`. This is maintained by NetBox and cannot be set directly. A field is available for use only while active; see [field status](../../customization/custom-fields.md#field-status).
+
 ### Label
 
 An optional human-friendly name for the custom field. If not defined, the field's `name` attribute will be used.

docs/release-notes/version-4.7.md

@@ -34,6 +34,8 @@
 * Webhooks now support a configurable timeout. If you have lowered `RQ_DEFAULT_TIMEOUT` to 60 seconds or less, you must also set [`WEBHOOK_DEFAULT_TIMEOUT`](../configuration/miscellaneous.md#webhook_default_timeout) to a lower value; NetBox will refuse to start otherwise.
 * Specifying an email server under the [`EMAIL`](../configuration/system.md#email) configuration parameter is now mandatory in order to send mail: A deployment which does not define `EMAIL['SERVER']` will raise an `InvalidMailer` exception when attempting to send, rather than failing at the SMTP connection.
 * The upgrade script now runs the `rebuild_config_context_cache` management command to populate the new config context cache. This may extend the duration of the upgrade for deployments with a large number of devices and virtual machines.
+* The obsolete `populate_custom_field_defaults()` method has been removed from `CustomFieldsMixin`.
+* `CustomField.objects.get_for_model()` and the `custom_fields` property of `CustomFieldsMixin` now return a list rather than a queryset, and `get_for_model()` returns only those fields which are active: Any whose stored data is being updated by a background job is omitted (see [field status](../customization/custom-fields.md#field-status)) unless selected via its `statuses` argument.
 * Removal of deprecated behavior
     * The `housekeeping` management command has been removed. (Its constituent tasks are performed by the individual management commands introduced in NetBox v4.6.)
     * NetBox's custom `querystring` template tag has been removed in favor of Django's built-in tag of the same name.

netbox/extras/api/serializers_/customfields.py

@@ -63,6 +63,9 @@ class CustomFieldSerializer(OwnerMixin, ChangeLogMessageSerializer, ValidatedMod
     )
     ui_visible = ChoiceField(choices=CustomFieldUIVisibleChoices, required=False)
     ui_editable = ChoiceField(choices=CustomFieldUIEditableChoices, required=False)
+    # A field is live only while active; the remaining states report a pending bulk update of its
+    # stored data. Read-only: the state is driven by the responsible background job.
+    status = ChoiceField(choices=CustomFieldStatusChoices, read_only=True)
 
     class Meta:
         model = CustomField
@@ -71,7 +74,7 @@ class CustomFieldSerializer(OwnerMixin, ChangeLogMessageSerializer, ValidatedMod
             'name', 'label', 'group_name', 'description', 'required', 'unique', 'search_weight', 'filter_logic',
             'ui_visible', 'ui_editable', 'is_cloneable', 'nulls_first', 'default', 'related_object_filter', 'weight',
             'validation_minimum', 'validation_maximum', 'validation_regex', 'validation_schema', 'choice_set',
-            'owner', 'comments', 'created', 'last_updated',
+            'status', 'owner', 'comments', 'created', 'last_updated',
         ]
         brief_fields = ('id', 'url', 'display', 'name', 'description')
 

netbox/extras/choices.py

@@ -47,6 +47,32 @@ class CustomFieldTypeChoices(ChoiceSet):
     )
 
 
+class CustomFieldStatusChoices(ChoiceSet):
+    """
+    The lifecycle state of a CustomField.
+
+    A field participates in object data only while active. The remaining states indicate that a bulk
+    update of its stored data is pending or in progress, during which the field is not live but its
+    row continues to reserve the field's name.
+    """
+    STATUS_ACTIVE = 'active'
+    STATUS_PROVISIONING = 'provisioning'
+    STATUS_DELETING = 'deleting'
+
+    CHOICES = (
+        (STATUS_ACTIVE, _('Active'), 'green'),
+        (STATUS_PROVISIONING, _('Provisioning'), 'cyan'),
+        (STATUS_DELETING, _('Deleting'), 'red'),
+    )
+
+    # The statuses in which the field's stored object data is its own: an active field's data is
+    # live, and a provisioning field's is being written by the job which will bring it live. Data
+    # held for a field in one of these statuses is left alone when an object is saved, and its
+    # default is populated on objects which lack it. A deleting field's data is on its way out, and
+    # so is excluded (see CustomFieldsMixin.clean() and get_defaults_for_model()).
+    DATA_STATUSES = (STATUS_ACTIVE, STATUS_PROVISIONING)
+
+
 class CustomFieldFilterLogicChoices(ChoiceSet):
 
     FILTER_DISABLED = 'disabled'

netbox/extras/constants.py

@@ -6,6 +6,15 @@ from extras.choices import LogLevelChoices
 # Custom fields
 CUSTOMFIELD_EMPTY_VALUES = (None, '', [])
 
+# Timeout (in seconds) applied to the background jobs which provision and purge custom field data.
+# These jobs exist precisely because the work is too large for the request which triggered it, so
+# the default RQ timeout -- being of the same order as the request timeout being escaped -- would
+# reimpose the limit they were introduced to avoid. A timeout is recoverable, as each job commits
+# its batches independently and both are idempotent, but it leaves the field pending until the job
+# is run again. Three hours is well beyond what a batched update of any real table takes, while
+# still releasing a worker blocked on an unresponsive database.
+CUSTOMFIELD_JOB_TIMEOUT = 10800
+
 # ImageAttachment
 IMAGE_ATTACHMENT_IMAGE_FORMATS = {
     'avif': 'image/avif',

netbox/extras/filtersets.py

@@ -187,7 +187,7 @@ class CustomFieldFilterSet(OwnerFilterMixin, ChangeLoggedModelFilterSet):
         fields = (
             'id', 'name', 'label', 'group_name', 'required', 'unique', 'search_weight', 'filter_logic', 'ui_visible',
             'ui_editable', 'weight', 'is_cloneable', 'nulls_first', 'description', 'validation_minimum',
-            'validation_maximum', 'validation_regex',
+            'validation_maximum', 'validation_regex', 'status',
         )
 
     def search(self, queryset, name, value):

netbox/extras/forms/filtersets.py

@@ -46,7 +46,9 @@ class CustomFieldFilterForm(OwnerFilterMixin, SavedFiltersMixin, FilterForm):
     model = CustomField
     fieldsets = (
         FieldSet('q', 'filter_id'),
-        FieldSet('object_type_id', 'type', 'group_name', 'weight', 'required', 'unique', name=_('Attributes')),
+        FieldSet(
+            'object_type_id', 'type', 'group_name', 'weight', 'required', 'unique', 'status', name=_('Attributes')
+        ),
         FieldSet('choice_set_id', 'related_object_type_id', name=_('Type Options')),
         FieldSet('ui_visible', 'ui_editable', 'is_cloneable', 'nulls_first', name=_('Behavior')),
         FieldSet('validation_minimum', 'validation_maximum', 'validation_regex', name=_('Validation')),
@@ -67,6 +69,11 @@ class CustomFieldFilterForm(OwnerFilterMixin, SavedFiltersMixin, FilterForm):
         required=False,
         label=_('Field type')
     )
+    status = forms.ChoiceField(
+        choices=add_blank_choice(CustomFieldStatusChoices),
+        required=False,
+        label=_('Status')
+    )
     group_name = forms.CharField(
         label=_('Group name'),
         required=False

netbox/extras/graphql/enums.py

@@ -10,6 +10,7 @@ __all__ = (
     'CustomFieldChoiceColorEnum',
     'CustomFieldChoiceSetBaseEnum',
     'CustomFieldFilterLogicEnum',
+    'CustomFieldStatusEnum',
     'CustomFieldTypeEnum',
     'CustomFieldUIEditableEnum',
     'CustomFieldUIVisibleEnum',
@@ -23,6 +24,7 @@ __all__ = (
 CustomFieldChoiceColorEnum = strawberry.enum(CustomFieldChoiceColorChoices.as_enum())
 CustomFieldChoiceSetBaseEnum = strawberry.enum(CustomFieldChoiceSetBaseChoices.as_enum())
 CustomFieldFilterLogicEnum = strawberry.enum(CustomFieldFilterLogicChoices.as_enum(prefix='filter'))
+CustomFieldStatusEnum = strawberry.enum(CustomFieldStatusChoices.as_enum(prefix='status'))
 CustomFieldTypeEnum = strawberry.enum(CustomFieldTypeChoices.as_enum(prefix='type'))
 CustomFieldUIEditableEnum = strawberry.enum(CustomFieldUIEditableChoices.as_enum())
 CustomFieldUIVisibleEnum = strawberry.enum(CustomFieldUIVisibleChoices.as_enum())

netbox/extras/graphql/filters.py

@@ -149,6 +149,9 @@ class CustomFieldFilter(ChangeLoggedModelFilter):
         strawberry_django.filter_field()
     )
     name: StrFilterLookup | None = strawberry_django.filter_field()
+    status: BaseFilterLookup[Annotated['CustomFieldStatusEnum', strawberry.lazy('extras.graphql.enums')]] | None = (
+        strawberry_django.filter_field()
+    )
     label: StrFilterLookup | None = strawberry_django.filter_field()
     group_name: StrFilterLookup | None = strawberry_django.filter_field()
     description: StrFilterLookup | None = strawberry_django.filter_field()

netbox/extras/jobs.py

@@ -5,9 +5,13 @@ 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 _
+from django_pg_utils import advisory_lock
 
 from core.signals import clear_events
 from dcim.models import Device
+from extras.choices import CustomFieldStatusChoices
+from extras.constants import CUSTOMFIELD_JOB_TIMEOUT
+from extras.models import CustomField
 from extras.models import Script as ScriptModel
 from netbox.context_managers import event_tracking
 from netbox.jobs import JobRunner
@@ -16,6 +20,21 @@ from utilities.exceptions import AbortScript, AbortTransaction
 
 from .utils import is_report
 
+__all__ = (
+    'CustomFieldDataJob',
+    'CustomFieldProvisioningJob',
+    'CustomFieldPurgeJob',
+    'RenderConfigContextJob',
+    'ScriptJob',
+    'provision_custom_field',
+    'purge_custom_field',
+)
+
+
+#
+# Config contexts
+#
+
 RENDER_CONFIG_CONTEXT_CHUNK_SIZE = 500
 
 # Safety bound on the number of re-scan passes performed by RenderConfigContextJob.run() (see the
@@ -108,6 +127,138 @@ class RenderConfigContextJob(JobRunner):
         return rendered
 
 
+#
+# Custom fields
+#
+
+
+def provision_custom_field(pk, object_type_pks):
+    """
+    Populate a new custom field's default value across the objects of the given types, then bring
+    the field live. Returns True if the field was brought live.
+
+    The backfill is committed in batches, so an interruption leaves the field provisioning with some
+    of its objects already updated. Running again completes it.
+
+    Args:
+        pk: The primary key of the CustomField to provision
+        object_type_pks: The primary keys of the object types to provision. Named explicitly, as
+            only the caller which deferred the work knows which of the field's assignments are the
+            new ones.
+    """
+    # Taken on the connection the field is written on, as CustomField.delete() takes it, so that
+    # the two are actually exclusive of one another.
+    using = router.db_for_write(CustomField)
+    with advisory_lock(CustomField.data_lock_key(pk), using=using):
+
+        # Rechecked now that the lock is held: where two jobs were enqueued for the same field,
+        # whichever arrived first has left it in a state the other no longer matches.
+        custom_field = CustomField.objects.filter(pk=pk, status=CustomFieldStatusChoices.STATUS_PROVISIONING).first()
+        if custom_field is None:
+            return False
+
+        # Restricted to the field's current assignments: a type unassigned since the job was
+        # enqueued must not be provisioned, its data having been removed by remove_data(). That
+        # method refuses an unassignment while the field is being provisioned, so this covers only
+        # a change made without it -- through the m2m table directly, which emits no signal.
+        object_types = custom_field.object_types.filter(pk__in=object_type_pks)
+        custom_field.populate_initial_data(object_types, commit_per_batch=True)
+
+        # Applied via the queryset so that bringing the field live does not record a change of its
+        # own, and cannot trip the guard in CustomField.clean().
+        activated = CustomField.objects.filter(
+            pk=pk, status=CustomFieldStatusChoices.STATUS_PROVISIONING
+        ).update(status=CustomFieldStatusChoices.STATUS_ACTIVE)
+        CustomField.objects.clear_cache()
+
+    return bool(activated)
+
+
+def purge_custom_field(pk):
+    """
+    Remove a deleted custom field's data from all applicable objects, then remove the field itself.
+    Returns True if the field was purged.
+
+    The row is dropped only once its data is gone: until then it reserves the field's name against a
+    new field which would otherwise inherit the orphaned values. The removal is committed in batches,
+    so an interruption leaves data behind for a later run to finish removing.
+
+    Args:
+        pk: The primary key of the CustomField to purge
+    """
+    # Taken on the connection the field is written on, as CustomField.delete() takes it, so that
+    # the two are actually exclusive of one another.
+    using = router.db_for_write(CustomField)
+    with advisory_lock(CustomField.data_lock_key(pk), using=using):
+
+        # Rechecked now that the lock is held: where two jobs were enqueued for the same field,
+        # whichever arrived first has left it in a state the other no longer matches.
+        custom_field = CustomField.objects.filter(pk=pk, status=CustomFieldStatusChoices.STATUS_DELETING).first()
+        if custom_field is None:
+            return False
+
+        custom_field.remove_stale_data(custom_field.object_types.all(), commit_per_batch=True)
+        custom_field._delete_row()
+
+    return True
+
+
+class CustomFieldDataJob(JobRunner):
+    """
+    Base class for the jobs which rewrite a custom field's stored data in bulk.
+
+    The field is passed by primary key rather than assigned to the job as its object. Job.clean()
+    permits only models with the jobs feature there, and granting CustomField that feature would
+    give it a cascading relation to its jobs -- so the purge job, whose last act is to remove the
+    row, would delete the record of its own execution as it ran.
+    """
+    @classmethod
+    def enqueue_for(cls, custom_field, **kwargs):
+        """
+        Enqueue this job for the given custom field, naming the field in the job's name and raising
+        its timeout from the default (see CUSTOMFIELD_JOB_TIMEOUT).
+        """
+        return cls.enqueue(
+            name=f'{cls.name}: {custom_field}',
+            custom_field_pk=custom_field.pk,
+            job_timeout=CUSTOMFIELD_JOB_TIMEOUT,
+            **kwargs,
+        )
+
+
+class CustomFieldProvisioningJob(CustomFieldDataJob):
+    """
+    Populate the default value of a newly created custom field.
+    """
+    class Meta:
+        name = 'Custom Field Provisioning'
+
+    def run(self, custom_field_pk, *args, object_type_pks, **kwargs):
+        if provision_custom_field(custom_field_pk, object_type_pks):
+            self.logger.info("Custom field provisioned")
+        else:
+            self.logger.info("Custom field is no longer awaiting provisioning; skipping")
+
+
+class CustomFieldPurgeJob(CustomFieldDataJob):
+    """
+    Purge the stored data of a deleted custom field, then delete the field.
+    """
+    class Meta:
+        name = 'Custom Field Purge'
+
+    def run(self, custom_field_pk, *args, **kwargs):
+        if purge_custom_field(custom_field_pk):
+            self.logger.info("Custom field data purged")
+        else:
+            self.logger.info("Custom field is no longer awaiting deletion; skipping")
+
+
+#
+# Scripts
+#
+
+
 class ScriptJob(JobRunner):
     """
     Script execution job.

netbox/extras/migrations/0144_customfield_status.py

@@ -0,0 +1,16 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('extras', '0143_event_rule_action_registry'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='customfield',
+            name='status',
+            field=models.CharField(default='active', editable=False, max_length=50),
+        ),
+    ]

netbox/extras/models/customfields.py

@@ -1,3 +1,4 @@
+import copy
 import decimal
 import json
 import re
@@ -8,8 +9,8 @@ import jsonschema
 from django import forms
 from django.conf import settings
 from django.core.validators import RegexValidator, ValidationError
-from django.db import models
-from django.db.models import F, Func, Value
+from django.db import connections, models, router, transaction
+from django.db.models import F, Func, Q, Value
 from django.urls import reverse
 from django.utils.html import escape
 from django.utils.safestring import mark_safe
@@ -20,6 +21,7 @@ from core.models import ObjectType
 from extras.choices import *
 from extras.data import CHOICE_SETS
 from extras.fields import ChoiceSetField
+from netbox.constants import ADVISORY_LOCK_KEYS
 from netbox.context import query_cache
 from netbox.models import ChangeLoggedModel
 from netbox.models.features import CloningMixin, ExportTemplatesMixin
@@ -27,6 +29,7 @@ from netbox.models.mixins import OwnerMixin
 from netbox.search import FieldTypes
 from utilities import filters
 from utilities.datetime import datetime_from_timestamp
+from utilities.exceptions import AbortRequest
 from utilities.forms.fields import (
     CSVChoiceField,
     CSVModelChoiceField,
@@ -65,38 +68,77 @@ SEARCH_TYPES = {
 class CustomFieldManager(models.Manager.from_queryset(RestrictedQuerySet)):
     use_in_migrations = True
 
-    def get_for_model(self, model):
+    def get_for_model(self, model, statuses=(CustomFieldStatusChoices.STATUS_ACTIVE,)):
         """
-        Return all CustomFields assigned to the given model.
+        Return a list of the CustomFields assigned to the given model which hold one of the given
+        statuses.
+
+        Only active fields are returned by default: a field awaiting a bulk update of its stored data
+        is not live, and must be invisible to every consumer of custom field data until that work
+        completes (see CustomFieldStatusChoices). This is the sole entry point by which custom fields
+        are resolved for an object, so excluding them here excludes them everywhere.
+
+        Every assigned field is fetched and cached whichever statuses are asked for, so that callers
+        wanting different subsets share one query per model per request.
+
+        Args:
+            model: The model whose custom fields are to be returned
+            statuses: The statuses to select (active only by default)
         """
-        # Check the request cache before hitting the database. Test the cached value against None
-        # rather than for truthiness: a model with no custom fields caches an empty QuerySet, which
-        # would otherwise be treated as a miss and re-queried on every call.
         cache = query_cache.get()
-        if cache is not None:
-            if (custom_fields := cache['custom_fields'].get(model._meta.model)) is not None:
-                return custom_fields
 
-        content_type = ObjectType.objects.get_for_model(model._meta.concrete_model)
-        custom_fields = self.get_queryset().filter(object_types=content_type).select_related(
-            'related_object_type', 'choice_set'
-        )
+        # Check the request cache before hitting the database. Test the cached value against None
+        # rather than for truthiness: a model with no custom fields caches an empty list, which
+        # would otherwise be treated as a miss and re-queried on every call.
+        custom_fields = cache['custom_fields'].get(model._meta.model) if cache is not None else None
+        if custom_fields is None:
+            content_type = ObjectType.objects.get_for_model(model._meta.concrete_model)
+            custom_fields = list(
+                self.get_queryset().filter(object_types=content_type).select_related(
+                    'related_object_type', 'choice_set'
+                )
+            )
 
-        # Populate the request cache to avoid redundant lookups
-        if cache is not None:
-            cache['custom_fields'][model._meta.model] = custom_fields
+            # Populate the request cache to avoid redundant lookups
+            if cache is not None:
+                cache['custom_fields'][model._meta.model] = custom_fields
 
-        return custom_fields
+        return [cf for cf in custom_fields if cf.status in statuses]
 
     def get_defaults_for_model(self, model):
         """
         Return a dictionary of serialized default values for all CustomFields applicable to the given model.
+
+        Fields still being provisioned are included, unlike in get_for_model(). The provisioning job
+        backfills only the objects which predate the field, so an object created while it runs must
+        pick up the default here or never receive one at all.
+
+        The defaults are assembled on each call from the fields cached by get_for_model() rather than
+        cached in their own right: building them costs a pass over a handful of objects already in
+        memory, where a second cache would have to be kept coherent with the first.
         """
-        custom_fields = self.get_for_model(model).filter(default__isnull=False)
+        custom_fields = self.get_for_model(model, statuses=CustomFieldStatusChoices.DATA_STATUSES)
+
+        # Copied so that a mutable default cannot be aliased into the object data of every object
+        # which takes it, the fields above being cached for the life of the request.
         return {
-            cf.name: cf.default for cf in custom_fields
+            cf.name: copy.deepcopy(cf.default) for cf in custom_fields if cf.default is not None
         }
 
+    @staticmethod
+    def clear_cache():
+        """
+        Discard the custom fields cached for the current request, so that a subsequent read reflects
+        a change which has been applied to the database without passing through save().
+
+        Called wherever a field's status is written directly (see CustomFieldStatusChoices): the
+        cache spans the whole of a request -- and the whole of a script or job run -- so a field
+        taken offline, brought live, or marked for deletion partway through one would otherwise
+        remain visible, or invisible, to everything which followed it there.
+        """
+        if (cache := query_cache.get()) is not None:
+            cache['custom_fields'].clear()
+
 
 class CustomField(CloningMixin, ExportTemplatesMixin, OwnerMixin, ChangeLoggedModel):
     object_types = models.ManyToManyField(
@@ -137,6 +179,14 @@ class CustomField(CloningMixin, ExportTemplatesMixin, OwnerMixin, ChangeLoggedMo
             ),
         )
     )
+    status = models.CharField(
+        max_length=50,
+        choices=CustomFieldStatusChoices,
+        default=CustomFieldStatusChoices.STATUS_ACTIVE,
+        verbose_name=_('status'),
+        help_text=_("Operational state of the field"),
+        editable=False
+    )
     label = models.CharField(
         verbose_name=_('label'),
         max_length=50,
@@ -315,6 +365,9 @@ class CustomField(CloningMixin, ExportTemplatesMixin, OwnerMixin, ChangeLoggedMo
             return self.choice_set.choices
         return []
 
+    def get_status_color(self):
+        return CustomFieldStatusChoices.colors.get(self.status)
+
     def get_ui_visible_color(self):
         return CustomFieldUIVisibleChoices.colors.get(self.ui_visible)
 
@@ -345,24 +398,212 @@ class CustomField(CloningMixin, ExportTemplatesMixin, OwnerMixin, ChangeLoggedMo
             return [{'value': v, 'label': self.get_choice_label(v)} for v in value]
         return value
 
-    def populate_initial_data(self, content_types):
+    @staticmethod
+    def data_lock_key(pk):
+        """
+        The advisory lock which serializes bulk updates of a field's stored data against one another
+        and against its deletion, keyed by primary key so that work on one field never waits on
+        another.
+        """
+        return ADVISORY_LOCK_KEYS['custom-field-data'], pk
+
+    @classmethod
+    def _try_lock_data(cls, pk, using):
+        """
+        Take the field's data lock at transaction scope, returning False if it is held elsewhere.
+        Never waits: a job holds this lock for the duration of its bulk update, which may run for
+        hours (see CUSTOMFIELD_JOB_TIMEOUT).
+        """
+        with connections[using].cursor() as cursor:
+            cursor.execute('SELECT pg_try_advisory_xact_lock(%s, %s)', cls.data_lock_key(pk))
+            return cursor.fetchone()[0]
+
+    def _lock_status(self, using):
+        """
+        Re-read the field's status under a row lock, returning None where the row no longer exists.
+
+        The status is not taken from this instance, which a job or a concurrent request may have
+        changed since it was fetched, and which must not change between being checked by the caller
+        and the field being marked below.
+        """
+        return self.__class__.objects.using(using).select_for_update().filter(
+            pk=self.pk
+        ).values_list('status', flat=True).first()
+
+    @staticmethod
+    def _update_object_data(model, filters=None, commit_per_batch=False, **update_kwargs):
+        """
+        Apply an UPDATE to the custom_field_data of every instance of the given model, in batches
+        of at most BULK_UPDATE_CHUNK_SIZE rows. Bounding the number of rows touched by each statement
+        keeps a very large table from exceeding the database statement timeout, as a JSONB update
+        rewrites each affected row in full.
+
+        :param filters: Optional Q object restricting which rows are updated. Negate it to address
+            the rows which do not match instead.
+        :param commit_per_batch: Commit each batch independently rather than wrapping them all in a
+            single transaction, so that a long-running job does not hold row locks for its whole
+            duration. Only for updates which can safely be resumed.
+        """
+        return chunked_update(
+            model.objects.filter(filters or Q()),
+            commit_per_batch=commit_per_batch,
+            **update_kwargs,
+        )
+
+    @staticmethod
+    def _exceeds_inline_limit(content_types):
+        """
+        Return True if a bulk update of custom field data across the given object types is too large
+        to perform within the request which triggered it, and must be handed to a background job
+        instead. The limit is BULK_UPDATE_CHUNK_SIZE objects across all of the given types: an
+        update which fits within a single statement is comfortably within any request timeout.
+
+        The rows are probed rather than counted: `COUNT(*)` reads the whole table, whereas counting
+        one primary key more than the limit costs the same on a table of ten million rows as on one
+        of ten thousand. Only the primary key is selected, and the model's default ordering cleared,
+        to keep the probe to an index-only scan.
+
+        On the deletion path this over-estimates, as every row of the type is counted where
+        remove_stale_data() would rewrite only those holding the field's key. Probing the key
+        instead would match the work exactly, but custom_field_data carries no index, so the LIMIT
+        could not bound the scan.
+        """
+        # Setting BULK_UPDATE_CHUNK_SIZE to None disables chunking, so the update would be issued
+        # as a single unbounded statement -- precisely what must not run inside a request. Treat any
+        # affected object as exceeding the limit, handing the work to the job, which issues that one
+        # statement under a timeout generous enough to survive it (see CUSTOMFIELD_JOB_TIMEOUT). A
+        # limit of zero leaves the probe below testing for a single row, so a field affecting no
+        # objects still needs no job.
+        limit = settings.BULK_UPDATE_CHUNK_SIZE
+        remaining = 0 if limit is None else limit
+
+        for ct in content_types:
+            if model := ct.model_class():
+                remaining -= model.objects.order_by().values_list('pk', flat=True)[:remaining + 1].count()
+                if remaining < 0:
+                    return True
+        return False
+
+    def provision_data(self, object_types):
+        """
+        Populate the field's default value across the existing objects of the given object types.
+
+        Where too many objects are affected to handle within the request, the field is taken offline
+        and the backfill handed to a background job: it does not go live until the job has finished
+        (see CustomFieldStatusChoices).
+
+        Assignment to a field which is not live is refused, as CustomField.clean() refuses every
+        other change to one: its configuration must not move under the job which is acting on it.
+        Were a second backfill deferred here, it would carry only the object types passed to it, and
+        whichever of the two jobs ran first would bring the field live -- leaving the other to find
+        a field it no longer matched, and its own object types silently unprovisioned.
+        """
+        from extras.jobs import CustomFieldProvisioningJob
+
+        using = router.db_for_write(self.__class__, instance=self)
+
+        with transaction.atomic(using=using):
+
+            # The status is re-read under a row lock rather than taken from this instance
+            self.status = self._lock_status(using)
+            if self.status is None:
+                # Deleted by a concurrent request since this instance was fetched; there is no field
+                # left to assign. Reported rather than ignored, as the assignment has not been applied.
+                raise AbortRequest(
+                    _("Custom field '{name}' no longer exists.").format(name=self.name)
+                )
+
+            if self.status != CustomFieldStatusChoices.STATUS_ACTIVE:
+                raise AbortRequest(
+                    _("Custom field '{name}' cannot be assigned to additional object types while its "
+                      "stored data is being updated (status: {status}).").format(
+                        name=self.name, status=self.get_status_display().lower()
+                    )
+                )
+
+            if self.default is None:
+                return
+
+            object_types = list(object_types)
+            if not self._exceeds_inline_limit(object_types):
+                self.populate_initial_data(object_types)
+                return
+
+            self.status = CustomFieldStatusChoices.STATUS_PROVISIONING
+            # Applied via the queryset so that taking the field offline does not itself record a change.
+            self.__class__.objects.using(using).filter(pk=self.pk).update(status=self.status)
+            self.__class__.objects.clear_cache()
+
+            # Deferred until commit so that the worker cannot observe the field before it is marked.
+            # The types are carried to the job, which cannot otherwise know which of the field's
+            # assignments are the new ones.
+            transaction.on_commit(
+                lambda: CustomFieldProvisioningJob.enqueue_for(
+                    self, object_type_pks=[ct.pk for ct in object_types]
+                ),
+                using=using
+            )
+
+    def remove_data(self, object_types):
+        """
+        Remove the field's stored data from the existing objects of the given object types, as the
+        field is unassigned from them.
+
+        Unassignment from a field which is not live is refused, as provision_data() refuses an
+        assignment to one. The job acting on the field's data carries the object types it was given
+        and would not observe an unassignment made under it: it would write its defaults into objects
+        the removal had already swept, then bring the field live with values left on objects it no
+        longer applies to.
+
+        Unlike provisioning and deletion, this is never deferred to a job. Only the objects which
+        actually hold a value for the field are rewritten, which on an unassignment is typically a
+        small fraction of the table (see the note in the custom fields documentation).
+        """
+        using = router.db_for_write(self.__class__, instance=self)
+
+        with transaction.atomic(using=using):
+
+            # The status is re-read under a row lock rather than taken from this instance, which a
+            # job may have taken offline since it was fetched, and which must not change between the
+            # check below and the data being removed.
+            self.status = self._lock_status(using)
+
+            if self.status is None:
+                # Deleted by a concurrent request since this instance was fetched; whatever data
+                # remains belongs to the deletion, which removes it in full.
+                raise AbortRequest(
+                    _("Custom field '{name}' no longer exists.").format(name=self.name)
+                )
+
+            if self.status != CustomFieldStatusChoices.STATUS_ACTIVE:
+                raise AbortRequest(
+                    _("Custom field '{name}' cannot be unassigned from object types while its "
+                      "stored data is being updated (status: {status}).").format(
+                        name=self.name, status=self.get_status_display().lower()
+                    )
+                )
+
+            self.remove_stale_data(object_types)
+
+    def populate_initial_data(self, content_types, commit_per_batch=False):
         """
         Populate initial custom field data upon either a) the creation of a new CustomField, or
         b) the assignment of an existing CustomField to new object types.
 
-        Only a non-null default is written. A field with no default has no value to record, and an
-        absent key is equivalent to a null one everywhere the data is read (see CustomFieldsMixin),
-        so materializing a JSON null on every object would be a very expensive no-op: on a large
-        table it can outlast the request. Objects without the key simply report no value until one
-        is assigned.
+        Objects which already hold a key for the field are left alone, making this idempotent -- as
+        a retried job requires, and as committing the backfill in batches relies on. (Note that a
+        cleared value is a JSON null rather than an absent key, and so is likewise preserved.)
         """
         if self.default is None:
             return
+
         value = Value(self.default, models.JSONField())
         for ct in content_types:
             if model := ct.model_class():
-                chunked_update(
-                    model.objects.all(),
+                self._update_object_data(
+                    model,
+                    filters=~Q(custom_field_data__has_key=self.name),
+                    commit_per_batch=commit_per_batch,
                     custom_field_data=Func(
                         F('custom_field_data'),
                         Value([self.name]),
@@ -371,19 +612,21 @@ class CustomField(CloningMixin, ExportTemplatesMixin, OwnerMixin, ChangeLoggedMo
                     )
                 )
 
-    def remove_stale_data(self, content_types):
+    def remove_stale_data(self, content_types, commit_per_batch=False):
         """
         Delete custom field data which is no longer relevant (either because the CustomField is
         no longer assigned to a model, or because it has been deleted).
 
-        Only objects which actually hold a value for the field are rewritten. Because keys are
-        materialized only when a value is set (see populate_initial_data()), this typically
-        excludes the bulk of the table.
+        Only objects which actually hold a value for the field are rewritten. That typically excludes
+        the bulk of the table, and makes this idempotent -- as committing the removal in batches
+        relies on -- since a row is dropped from the queryset by the update which removes its key.
         """
         for ct in content_types:
             if model := ct.model_class():
-                chunked_update(
-                    model.objects.filter(custom_field_data__has_key=self.name),
+                self._update_object_data(
+                    model,
+                    filters=Q(custom_field_data__has_key=self.name),
+                    commit_per_batch=commit_per_batch,
                     custom_field_data=F('custom_field_data') - self.name
                 )
 
@@ -394,8 +637,9 @@ class CustomField(CloningMixin, ExportTemplatesMixin, OwnerMixin, ChangeLoggedMo
         """
         for ct in self.object_types.all():
             if model := ct.model_class():
-                chunked_update(
-                    model.objects.filter(custom_field_data__has_key=old_name),
+                self._update_object_data(
+                    model,
+                    filters=Q(custom_field_data__has_key=old_name),
                     custom_field_data=Func(
                         F('custom_field_data') - old_name,
                         Value([new_name]),
@@ -408,9 +652,95 @@ class CustomField(CloningMixin, ExportTemplatesMixin, OwnerMixin, ChangeLoggedMo
                         function='jsonb_set')
                 )
 
+    def delete(self, using=None, *args, **kwargs):
+        """
+        Delete the field, deferring the removal of its stored data to a background job where too
+        many objects are affected to handle within the request (see #22996).
+
+        Where the work is deferred, the row is retained until the job completes: `name` is unique, so
+        for as long as the row exists no other field can take this name and inherit the data still
+        awaiting removal.
+
+        The deletion signals are dispatched here rather than when the row is finally removed, so that
+        protection rules, the change log, event rules and the search index observe the deletion where
+        the user performed it. They run again in the worker, where every effect beyond the protection
+        rules is gated on there being a current request, making the replay a no-op.
+
+        The deletion is refused outright if a background job holds the field's data lock, rather than
+        queueing behind that job. This applies equally to a field already pending deletion: reporting
+        a deletion which did not happen would be worse than refusing it. A field stranded in a pending
+        state by a job which never ran holds no lock, and stays deletable; retrying the deletion of
+        one already pending enqueues a fresh purge job for it.
+
+        Deleting a field already marked for deletion -- by an earlier request of the user's own, or by
+        a concurrent one -- removes nothing further and dispatches no second set of deletion signals.
+        """
+        from extras.jobs import CustomFieldPurgeJob
+
+        using = using or router.db_for_write(self.__class__, instance=self)
+
+        with transaction.atomic(using=using):
+            if not self._try_lock_data(self.pk, using):
+                raise AbortRequest(
+                    _("Custom field '{name}' is being updated by a background job and cannot be "
+                      "deleted until that job has completed.").format(name=self.name)
+                )
+
+            # The status is re-read under a row lock rather than taken from this instance
+            self.status = self._lock_status(using)
+            if self.status is None:
+                # Already deleted outright by a concurrent request; nothing remains to delete.
+                return 0, {}
+
+            if self.status == CustomFieldStatusChoices.STATUS_DELETING:
+                # Already pending deletion; the purge job will remove the row once its data is gone.
+                # The lock being free, no job is *running*, so the one enqueued when the field was
+                # marked may never have run: enqueue another, delete() being the only route to one.
+                # Left as it is, a field whose job never ran could never be removed, and would hold
+                # its name against a replacement indefinitely. Where that job is merely queued (a
+                # concurrent deletion having just marked the field), the second job is harmless:
+                # purge_custom_field() rechecks the status under the lock and no-ops.
+                transaction.on_commit(lambda: CustomFieldPurgeJob.enqueue_for(self), using=using)
+                return 0, {}
+
+            if not self._exceeds_inline_limit(self.object_types.all()):
+                # Few enough objects to purge within the request: delete the row outright, its
+                # stored data being removed by handle_cf_deleted().
+                return super().delete(using, *args, **kwargs)
+
+            # Update the custom field's status before the signals are dispatched. Applied via the
+            # queryset to avoid emitting a spurious "updated" change record.
+            self.status = CustomFieldStatusChoices.STATUS_DELETING
+            self.__class__.objects.using(using).filter(pk=self.pk).update(status=self.status)
+            self.__class__.objects.clear_cache()
+
+            models.signals.pre_delete.send(sender=self.__class__, instance=self, using=using, origin=self)
+            models.signals.post_delete.send(sender=self.__class__, instance=self, using=using, origin=self)
+
+            # Deferred until commit so that the worker cannot observe the field before it is marked,
+            # and is not enqueued at all if the deletion is aborted.
+            transaction.on_commit(lambda: CustomFieldPurgeJob.enqueue_for(self), using=using)
+
+        return 1, {self._meta.label: 1}
+
+    def _delete_row(self):
+        """
+        Remove the row itself. Called by CustomFieldPurgeJob once the field's stored data has been
+        purged; nothing else should bypass delete().
+        """
+        return super().delete()
+
     def clean(self):
         super().clean()
 
+        # A field awaiting a bulk update of its stored data is not live, and its configuration must
+        # not change under the job which is acting on it.
+        if self.pk and self.status != CustomFieldStatusChoices.STATUS_ACTIVE:
+            raise ValidationError(
+                _("Custom field '{name}' cannot be modified while its stored data is being updated "
+                  "(status: {status}).").format(name=self.name, status=self.get_status_display().lower())
+            )
+
         # Validate the field's default value (if any)
         if self.default is not None:
             try:

netbox/extras/signals.py

@@ -1,9 +1,10 @@
 from django.contrib.contenttypes.models import ContentType
-from django.db.models.signals import m2m_changed, post_save, pre_delete
+from django.db.models.signals import m2m_changed, post_delete, post_save, pre_delete
 from django.dispatch import receiver
 
 from core.events import *
 from core.signals import job_end, job_start
+from extras.choices import CustomFieldStatusChoices
 from extras.events import EventContext, process_event_rules
 from extras.models import EventRule, Notification, Subscription
 from netbox.config import get_config
@@ -33,26 +34,31 @@ def handle_cf_object_types_changed(instance, action, pk_set, reverse, **kwargs):
     Only the forward direction is handled: every action below operates on the CustomField, whereas
     the reverse of this relation (ContentType.custom_fields) reports the ContentType as the sender's
     instance. Nothing in NetBox assigns object types that way.
+
+    Both unassignment actions are handled before the fact, so that remove_data() refusing the change
+    precedes the removal of the assignments themselves. Django wraps each of these operations in a
+    transaction, so the refusal would roll the removal back in any case -- but only where the caller
+    left that transaction to it.
     """
-    if reverse or action not in ('pre_clear', 'post_add', 'post_remove'):
+    if reverse or action not in ('pre_clear', 'post_add', 'pre_remove'):
         return
 
     if action == 'pre_clear':
-        # clear() unassigns every object type at once. It must be handled before the fact: no
-        # pk_set is reported for a clear, so the assignments have to be read while they still
-        # exist. (Note that set() diffs via remove()/add() by default, so it does not land here.)
-        instance.remove_stale_data(instance.object_types.all())
+        # clear() unassigns every object type at once, and reports no pk_set, so the assignments
+        # have to be read while they still exist. (Note that set() diffs via remove()/add() by
+        # default, so it does not land here.)
+        instance.remove_data(instance.object_types.all())
         return
 
     object_types = ContentType.objects.filter(pk__in=pk_set)
 
     if action == 'post_add':
-        # Populate the field's default value (if any) on all existing objects
-        instance.populate_initial_data(object_types)
-
+        # Populate the field's default value (if any) on the existing objects of the types just
+        # assigned.
+        instance.provision_data(object_types)
     else:
-        # Remove the field's stored data from objects to which it no longer applies
-        instance.remove_stale_data(object_types)
+        # Remove the field's stored data from objects to which it no longer applies.
+        instance.remove_data(object_types)
 
 
 def handle_cf_renamed(instance, created, **kwargs):
@@ -66,14 +72,41 @@ def handle_cf_renamed(instance, created, **kwargs):
 def handle_cf_deleted(instance, **kwargs):
     """
     Handle the cleanup of old custom field data when a CustomField is deleted.
+
+    A field already marked for deletion is skipped: its data is too voluminous to purge inline, and
+    CustomFieldPurgeJob is removing it (see CustomField.delete()).
     """
-    instance.remove_stale_data(instance.object_types.all())
+    if instance.status != CustomFieldStatusChoices.STATUS_DELETING:
+        instance.remove_stale_data(instance.object_types.all())
+
+
+def handle_cf_cache_invalidation(action=None, **kwargs):
+    """
+    Discard the custom fields cached for the current request whenever one is created, modified,
+    deleted, or (un)assigned from an object type.
+
+    The cache spans the whole of a request -- and the whole of a script or job run, which share one
+    for their entire duration -- so without this a field created or changed partway through would be
+    served from what was read before it, to everything which followed.
+
+    A field's status is written via the queryset and so reaches none of these signals; the paths
+    which write it clear the cache themselves (see CustomFieldManager.clear_cache).
+    """
+    # m2m_changed fires either side of the change; clear once it has actually been applied.
+    if action is not None and not action.startswith('post_'):
+        return
+
+    CustomField.objects.clear_cache()
 
 
 post_save.connect(handle_cf_renamed, sender=CustomField)
 pre_delete.connect(handle_cf_deleted, sender=CustomField)
 m2m_changed.connect(handle_cf_object_types_changed, sender=CustomField.object_types.through)
 
+post_save.connect(handle_cf_cache_invalidation, sender=CustomField)
+post_delete.connect(handle_cf_cache_invalidation, sender=CustomField)
+m2m_changed.connect(handle_cf_cache_invalidation, sender=CustomField.object_types.through)
+
 
 #
 # Custom validation

netbox/extras/tables/columns.py

@@ -1,12 +1,40 @@
+import django_tables2 as tables
+from django.utils.html import format_html
 from django.utils.translation import gettext as _
 
+from extras.choices import CustomFieldStatusChoices
 from netbox.tables.columns import ActionsColumn, ActionsItem
 
 __all__ = (
+    'CustomFieldStatusColumn',
     'NotificationActionsColumn',
 )
 
 
+class CustomFieldStatusColumn(tables.Column):
+    """
+    Render a custom field's status as an icon: a checkmark where the field is live, and a warning
+    where a bulk update of its stored data is still pending (see CustomFieldStatusChoices).
+
+    An icon because the status is worth noting only in the exceptional case, which is any state
+    other than active. The full label is given as hover text, and is what an export records.
+    """
+    ICONS = {
+        True: ('text-bg-green', 'mdi-check-bold'),
+        False: ('text-bg-orange', 'mdi-alert'),
+    }
+
+    def render(self, record):
+        css_class, icon = self.ICONS[record.status == CustomFieldStatusChoices.STATUS_ACTIVE]
+        return format_html(
+            '<span class="badge {}" title="{}"><i class="mdi {}"></i></span>',
+            css_class, record.get_status_display(), icon
+        )
+
+    def value(self, record):
+        return record.get_status_display()
+
+
 class NotificationActionsColumn(ActionsColumn):
     actions = {
         'dismiss': ActionsItem(_('Dismiss'), 'trash-can-outline', 'delete', 'danger'),

netbox/extras/tables/tables.py

@@ -12,7 +12,7 @@ from netbox.constants import EMPTY_TABLE_TEXT
 from netbox.events import get_event_text
 from netbox.tables import BaseTable, NetBoxTable, PrimaryModelTable, columns
 
-from .columns import NotificationActionsColumn
+from .columns import CustomFieldStatusColumn, NotificationActionsColumn
 
 __all__ = (
     'BookmarkTable',
@@ -87,6 +87,9 @@ class CustomFieldTable(NetBoxTable):
         verbose_name=_('Validate Uniqueness'),
         false_mark=None
     )
+    status = CustomFieldStatusColumn(
+        verbose_name=_('Status')
+    )
     ui_visible = columns.ChoiceFieldColumn(
         verbose_name=_('Visible')
     )
@@ -140,10 +143,12 @@ class CustomFieldTable(NetBoxTable):
             'pk', 'id', 'name', 'object_types', 'label', 'type', 'related_object_type', 'group_name', 'required',
             'unique', 'default', 'description', 'search_weight', 'filter_logic', 'ui_visible', 'ui_editable',
             'is_cloneable', 'nulls_first', 'weight', 'choice_set', 'choices', 'validation_minimum',
-            'validation_maximum', 'validation_regex', 'validation_schema', 'comments', 'created', 'last_updated',
+            'validation_maximum', 'validation_regex', 'validation_schema', 'status', 'comments', 'created',
+            'last_updated',
         )
         default_columns = (
-            'pk', 'name', 'object_types', 'label', 'group_name', 'type', 'required', 'unique', 'description',
+            'pk', 'name', 'status', 'object_types', 'label', 'group_name', 'type', 'required', 'unique',
+            'description',
         )
 
 

netbox/extras/tests/test_filtersets.py

@@ -143,6 +143,12 @@ class CustomFieldTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
         params = {'ui_editable': CustomFieldUIEditableChoices.YES}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
 
+    def test_status(self):
+        params = {'status': CustomFieldStatusChoices.STATUS_ACTIVE}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 6)
+        params = {'status': CustomFieldStatusChoices.STATUS_DELETING}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 0)
+
     def test_choice_set(self):
         params = {'choice_set': ['Choice Set 1', 'Choice Set 2']}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)

netbox/extras/tests/test_graphql.py

@@ -6,9 +6,9 @@ from rest_framework import status
 from core.events import OBJECT_CREATED
 from core.models import ObjectType
 from dcim.models import Site
-from extras.choices import EventRuleActionChoices
+from extras.choices import CustomFieldStatusChoices, CustomFieldTypeChoices, EventRuleActionChoices
 from extras.graphql.enums import EventRuleActionEnum
-from extras.models import EventRule, Webhook
+from extras.models import CustomField, EventRule, Webhook
 from utilities.testing import APITestCase
 
 
@@ -54,3 +54,31 @@ class EventRuleActionEnumTestCase(APITestCase):
         self.assertNotIn('errors', data)
         names = {rule['name'] for rule in data['data']['event_rule_list']}
         self.assertEqual(names, {'GraphQL Enum Webhook Rule'})
+
+
+class CustomFieldStatusFilterTestCase(APITestCase):
+    """A field which is not live is invisible everywhere else, so its status must be queryable."""
+
+    def test_filter_custom_fields_by_status(self):
+        site_type = ObjectType.objects.get_for_model(Site)
+        for name, status_ in (
+            ('graphql_active_field', CustomFieldStatusChoices.STATUS_ACTIVE),
+            ('graphql_provisioning_field', CustomFieldStatusChoices.STATUS_PROVISIONING),
+        ):
+            custom_field = CustomField.objects.create(type=CustomFieldTypeChoices.TYPE_TEXT, name=name)
+            custom_field.object_types.set([site_type])
+            # Applied via the queryset, as CustomField.status is not directly writable
+            CustomField.objects.filter(pk=custom_field.pk).update(status=status_)
+
+        self.add_permissions('extras.view_customfield')
+        url = reverse('graphql')
+        query = '{custom_field_list(filters: {status: {exact: STATUS_PROVISIONING}}) {name status}}'
+        response = self.client.post(url, data={'query': query}, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+
+        data = json.loads(response.content)
+        self.assertNotIn('errors', data)
+        self.assertEqual(
+            [(cf['name'], cf['status']) for cf in data['data']['custom_field_list']],
+            [('graphql_provisioning_field', CustomFieldStatusChoices.STATUS_PROVISIONING)]
+        )

netbox/extras/tests/test_tables.py

@@ -3,7 +3,8 @@ from django.test import TestCase
 from core.events import OBJECT_CREATED
 from core.models import ObjectType
 from dcim.models import Site
-from extras.models import Bookmark, EventRule, Notification, Subscription
+from extras.choices import CustomFieldStatusChoices, CustomFieldTypeChoices
+from extras.models import Bookmark, CustomField, EventRule, Notification, Subscription
 from extras.tables import *
 from utilities.testing import TableTestCases
 
@@ -12,6 +13,66 @@ class CustomFieldTableTestCase(TableTestCases.StandardTableTestCase):
     table = CustomFieldTable
 
 
+class CustomFieldStatusColumnTestCase(TestCase):
+    """
+    A field which is not live must be distinguishable at a glance from one which is: deleting a
+    field with a large amount of stored data reports success while leaving it listed until the purge
+    job completes (see CustomFieldStatusColumn).
+    """
+    @classmethod
+    def setUpTestData(cls):
+        for status in (
+            CustomFieldStatusChoices.STATUS_ACTIVE,
+            CustomFieldStatusChoices.STATUS_PROVISIONING,
+            CustomFieldStatusChoices.STATUS_DELETING,
+        ):
+            custom_field = CustomField.objects.create(
+                name=f'field_{status}', type=CustomFieldTypeChoices.TYPE_TEXT
+            )
+            # Applied via the queryset to bypass the guard against modifying a pending field
+            CustomField.objects.filter(pk=custom_field.pk).update(status=status)
+
+    def _row(self, status):
+        table = CustomFieldTable(CustomField.objects.filter(status=status))
+        return table.rows[0]
+
+    def test_status_is_shown_by_default(self):
+        self.assertIn('status', CustomFieldTable.Meta.default_columns)
+
+    def test_active_field_renders_a_green_checkmark(self):
+        cell = self._row(CustomFieldStatusChoices.STATUS_ACTIVE).get_cell('status')
+
+        self.assertInHTML(
+            '<span class="badge text-bg-green" title="Active"><i class="mdi mdi-check-bold"></i></span>', cell
+        )
+
+    def test_pending_field_renders_an_orange_warning(self):
+        for status, label in (
+            (CustomFieldStatusChoices.STATUS_PROVISIONING, 'Provisioning'),
+            (CustomFieldStatusChoices.STATUS_DELETING, 'Deleting'),
+        ):
+            with self.subTest(status=status):
+                cell = self._row(status).get_cell('status')
+
+                self.assertInHTML(
+                    f'<span class="badge text-bg-orange" title="{label}">'
+                    f'<i class="mdi mdi-alert"></i></span>',
+                    cell
+                )
+
+    def test_export_records_the_label(self):
+        """
+        The icon carries no text, so an export must fall back to the human-readable status.
+        """
+        for status, label in (
+            (CustomFieldStatusChoices.STATUS_ACTIVE, 'Active'),
+            (CustomFieldStatusChoices.STATUS_PROVISIONING, 'Provisioning'),
+            (CustomFieldStatusChoices.STATUS_DELETING, 'Deleting'),
+        ):
+            with self.subTest(status=status):
+                self.assertEqual(self._row(status).get_cell_value('status'), label)
+
+
 class CustomFieldChoiceSetTableTestCase(TableTestCases.StandardTableTestCase):
     table = CustomFieldChoiceSetTable
 

netbox/extras/ui/panels.py

@@ -126,6 +126,7 @@ class CustomFieldPanel(panels.ObjectAttributesPanel):
     title = _('Custom Field')
 
     name = attrs.TextAttr('name')
+    status = attrs.ChoiceAttr('status')
     type = attrs.TemplatedAttr('type', label=_('Type'), template_name='extras/customfield/attrs/type.html')
     label = attrs.TextAttr('label')
     group_name = attrs.TextAttr('group_name', label=_('Group name'))

netbox/netbox/constants.py

@@ -31,6 +31,9 @@ ADVISORY_LOCK_KEYS = {
 
     # Jobs
     'job-schedules': 110100,
+
+    # Custom field data
+    'custom-field-data': 115100,
 }
 
 # General-purpose tokens

netbox/netbox/models/features.py

@@ -222,12 +222,12 @@ class CustomFieldsMixin(models.Model):
     @cached_property
     def custom_fields(self):
         """
-        Return the QuerySet of CustomFields assigned to this model.
+        Return the list of CustomFields assigned to this model.
 
         ```python
         >>> tenant = Tenant.objects.first()
         >>> tenant.custom_fields
-        <RestrictedQuerySet [<CustomField: Primary site>, <CustomField: Customer ID>, <CustomField: Is active>]>
+        [<CustomField: Primary site>, <CustomField: Customer ID>, <CustomField: Is active>]
         ```
         """
         from extras.models import CustomField
@@ -277,9 +277,10 @@ class CustomFieldsMixin(models.Model):
         """
         from extras.models import CustomField
         groups = defaultdict(dict)
-        visible_custom_fields = CustomField.objects.get_for_model(self).exclude(
-            ui_visible=CustomFieldUIVisibleChoices.HIDDEN
-        )
+        visible_custom_fields = [
+            cf for cf in CustomField.objects.get_for_model(self)
+            if cf.ui_visible != CustomFieldUIVisibleChoices.HIDDEN
+        ]
 
         for cf in visible_custom_fields:
             value = self.custom_field_data.get(cf.name)
@@ -290,38 +291,43 @@ class CustomFieldsMixin(models.Model):
 
         return dict(groups)
 
-    def populate_custom_field_defaults(self):
-        """
-        Apply the default value for each custom field
-        """
-        for cf in self.custom_fields:
-            self.custom_field_data[cf.name] = cf.default
-    populate_custom_field_defaults.alters_data = True
-
     def clean(self):
         super().clean()
         from extras.models import CustomField
 
+        # Fields still being provisioned are fetched alongside the active ones, but are not live:
+        # their stored data belongs to the job acting on it, so it is neither validated below nor
+        # pruned as stale -- while remaining subject to the defaults applied in save(), which draws
+        # on this same set of statuses. Only active fields are validated or enforced as required.
+        assigned_fields = CustomField.objects.get_for_model(
+            self, statuses=CustomFieldStatusChoices.DATA_STATUSES
+        )
         custom_fields = {
-            cf.name: cf for cf in CustomField.objects.get_for_model(self)
+            cf.name: cf for cf in assigned_fields
+            if cf.status == CustomFieldStatusChoices.STATUS_ACTIVE
         }
 
         # Remove any stale custom field data
+        assigned_names = {cf.name for cf in assigned_fields}
         self.custom_field_data = {
-            k: v for k, v in self.custom_field_data.items() if k in custom_fields.keys()
+            k: v for k, v in self.custom_field_data.items() if k in assigned_names
         }
 
         # Validate all field values
         for field_name, value in self.custom_field_data.items():
+            if (cf := custom_fields.get(field_name)) is None:
+                # The field is not live; its value is left to the job which is provisioning it
+                continue
+
             try:
-                custom_fields[field_name].validate(value)
+                cf.validate(value)
             except ValidationError as e:
                 raise ValidationError(_("Invalid value for custom field '{name}': {error}").format(
                     name=field_name, error=e.message
                 ))
 
             # Validate uniqueness if enforced
-            if custom_fields[field_name].unique and value not in CUSTOMFIELD_EMPTY_VALUES:
+            if cf.unique and value not in CUSTOMFIELD_EMPTY_VALUES:
                 if self._meta.model.objects.exclude(pk=self.pk).filter(**{
                     f'custom_field_data__{field_name}': value
                 }).exists():
@@ -337,10 +343,13 @@ class CustomFieldsMixin(models.Model):
     def save(self, *args, **kwargs):
         from extras.models import CustomField
 
-        # Populate default values for custom fields not already present in the object data
-        for cf in CustomField.objects.get_for_model(self):
-            if cf.name not in self.custom_field_data and cf.default is not None:
-                self.custom_field_data[cf.name] = cf.default
+        # Populate default values for custom fields not already present in the object data. This
+        # covers fields still being provisioned as well as active ones, so that an object created
+        # while a new field is being backfilled does not miss its default (see
+        # CustomFieldManager.get_defaults_for_model()).
+        for name, default in CustomField.objects.get_defaults_for_model(self).items():
+            if name not in self.custom_field_data:
+                self.custom_field_data[name] = default
 
         super().save(*args, **kwargs)
 

netbox/utilities/querysets.py

@@ -1,3 +1,5 @@
+from contextlib import nullcontext
+
 from django.conf import settings
 from django.db import router, transaction
 from django.db.models import Max, Prefetch, QuerySet
@@ -12,7 +14,7 @@ __all__ = (
 )
 
 
-def chunked_update(queryset, chunk_size=None, **kwargs):
+def chunked_update(queryset, chunk_size=None, commit_per_batch=False, **kwargs):
     """
     Perform a bulk UPDATE on the given queryset, optionally splitting it into batches of at most
     `chunk_size` rows. Bounding the number of rows touched by each statement avoids exceeding the
@@ -28,6 +30,17 @@ def chunked_update(queryset, chunk_size=None, **kwargs):
     :param queryset: The QuerySet identifying the rows to update
     :param chunk_size: The maximum number of rows to update per statement (defaults to
         settings.BULK_UPDATE_CHUNK_SIZE)
+    :param commit_per_batch: Commit each batch independently rather than wrapping them all in a
+        single transaction, forfeiting the atomicity described above. Postgres holds a row lock on
+        every row updated until the transaction commits, which for a long-running job spanning a
+        large table means blocking concurrent edits for its whole run, so such callers commit as
+        they go. Only for updates which can safely be resumed, and only outside an enclosing atomic
+        block, which owns the commit regardless -- passed from within one, this silently degrades to
+        a single transaction rather than raising, as Django's TestCase wraps every test in one.
+
+        The callers which pass it (see extras.jobs) run in a worker, under a session-scoped advisory
+        lock which is taken through a bare cursor, so nothing in that path opens a transaction and
+        each batch does commit as intended.
     """
     if chunk_size is None:
         chunk_size = settings.BULK_UPDATE_CHUNK_SIZE
@@ -48,7 +61,7 @@ def chunked_update(queryset, chunk_size=None, **kwargs):
     # Upper bound on the PKs to process. Established lazily (see below) only once a second batch is
     # known to be needed, so the common single-batch case incurs no extra aggregate query.
     max_pk = None
-    with transaction.atomic(using=using):
+    with nullcontext() if commit_per_batch else transaction.atomic(using=using):
         while True:
             batch = queryset.using(using).filter(pk__gt=last_pk).order_by('pk')
             if max_pk is not None: