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

docs/customization/custom-fields.md

@@ -37,7 +37,28 @@ 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 which holds data, both require rewriting the stored data of every object the field applies to. Where a large number of objects is affected, 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. |
+
+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 data is neither read nor written; 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 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 (`rqworker`). A field left mid-operation, for example because no worker was running, is picked up by NetBox's daily housekeeping job.
+
+!!! 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.

netbox/core/jobs.py

@@ -10,6 +10,9 @@ from django.utils import timezone
 from packaging import version
 
 from core.models import Job, ObjectChange
+from extras.choices import CustomFieldStatusChoices
+from extras.jobs import CustomFieldProvisioningJob, CustomFieldPurgeJob
+from extras.models import CustomField
 from netbox.config import Config
 from netbox.jobs import JobRunner, system_job
 from netbox.search.backends import search_backend
@@ -79,6 +82,7 @@ class SystemHousekeepingJob(JobRunner):
         self.clear_expired_sessions()
         self.prune_changelog()
         self.delete_expired_jobs()
+        self.finalize_custom_fields()
         self.check_for_new_releases()
 
     def send_census_report(self):
@@ -191,6 +195,26 @@ class SystemHousekeepingJob(JobRunner):
         count = Job.objects.filter(created__lt=cutoff).delete()[0]
         self.logger.info(f"Deleted {count} expired jobs")
 
+    def finalize_custom_fields(self):
+        """
+        Complete any pending custom field data operations.
+
+        Provisioning and purging are ordinarily performed by a dedicated job enqueued at the time
+        the field is created or deleted. This is a backstop for the cases where that job never ran
+        or failed: a field left mid-operation is not live, and a field left pending deletion holds
+        its name reserved, so neither can be allowed to persist indefinitely.
+        """
+        self.logger.info("Finalizing pending custom fields...")
+        jobs = {
+            CustomFieldStatusChoices.STATUS_PROVISIONING: CustomFieldProvisioningJob,
+            CustomFieldStatusChoices.STATUS_DELETING: CustomFieldPurgeJob,
+        }
+        count = 0
+        for custom_field in CustomField.objects.filter(status__in=jobs):
+            jobs[custom_field.status].enqueue_for(custom_field, skip_locked=True)
+            count += 1
+        self.logger.info(f"Enqueued {count} custom field jobs.")
+
     def check_for_new_releases(self):
         """
         Check for new releases and cache the latest release.

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,25 @@ 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'),
+    )
+
+
 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
+# housekeeping backstop next runs. 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/jobs.py

@@ -5,9 +5,14 @@ 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.models import ObjectType
 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 +21,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 +128,147 @@ class RenderConfigContextJob(JobRunner):
         return rendered
 
 
+#
+# Custom fields
+#
+
+
+def provision_custom_field(pk, object_type_pks=None, skip_locked=False):
+    """
+    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 provisioned.
+
+    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, or None to provision
+            every type currently assigned to the field. The housekeeping backstop passes None, as it
+            has no record of which assignments deferred the work; that provisions more types than the
+            deferred job would have, but is preferable to leaving the field offline indefinitely.
+        skip_locked: Return False rather than waiting if the field's data lock is already held by
+            another caller
+    """
+    # 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), wait=not skip_locked, using=using) as acquired:
+        if not acquired:
+            return False
+
+        # Rechecked now that the lock is held: whichever of this job and the housekeeping backstop
+        # arrived first has left the field 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
+
+        if object_type_pks is None:
+            object_types = custom_field.object_types.all()
+        else:
+            object_types = ObjectType.objects.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().
+        CustomField.objects.filter(pk=pk).update(status=CustomFieldStatusChoices.STATUS_ACTIVE)
+
+    return True
+
+
+def purge_custom_field(pk, skip_locked=False):
+    """
+    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
+        skip_locked: Return False rather than waiting if the field's data lock is already held by
+            another caller
+    """
+    # 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), wait=not skip_locked, using=using) as acquired:
+        if not acquired:
+            return False
+
+        # Rechecked now that the lock is held: whichever of this job and the housekeeping backstop
+        # arrived first has left the field 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.
+
+    skip_locked is set where the job was enqueued by the housekeeping backstop, for which the field
+    is only a candidate: the job properly responsible for it may still be working through it, and
+    must be left to finish rather than blocking a worker here for its duration.
+    """
+    @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=None, skip_locked=False, **kwargs):
+        if provision_custom_field(custom_field_pk, object_type_pks, skip_locked=skip_locked):
+            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, skip_locked=False, **kwargs):
+        if purge_custom_field(custom_field_pk, skip_locked=skip_locked):
+            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

@@ -8,18 +8,20 @@ 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 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
 from django.utils.translation import gettext_lazy as _
+from django_pg_utils import advisory_lock
 from jsonschema.exceptions import ValidationError as JSONValidationError
 
 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,72 @@ 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.
         """
-        custom_fields = self.get_for_model(model).filter(default__isnull=False)
-        return {
-            cf.name: cf.default for cf in custom_fields
+        # Check the request cache before hitting the database. Test the cached value against None
+        # rather than for truthiness, as a model with no defaults caches an empty dict.
+        cache = query_cache.get()
+        if cache is not None:
+            if (defaults := cache['custom_field_defaults'].get(model._meta.model)) is not None:
+                return defaults
+
+        custom_fields = self.get_for_model(model, statuses=(
+            CustomFieldStatusChoices.STATUS_ACTIVE,
+            CustomFieldStatusChoices.STATUS_PROVISIONING,
+        ))
+        defaults = {
+            cf.name: cf.default for cf in custom_fields if cf.default is not None
         }
 
+        # Populate the request cache to avoid redundant lookups
+        if cache is not None:
+            cache['custom_field_defaults'][model._meta.model] = defaults
+
+        return defaults
+
 
 class CustomField(CloningMixin, ExportTemplatesMixin, OwnerMixin, ChangeLoggedModel):
     object_types = models.ManyToManyField(
@@ -137,6 +174,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 +360,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 +393,120 @@ 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):
         """
-        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.
+        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.
+
+        An advisory lock rather than a lock on the field's own row, because the jobs commit their
+        batches independently: a row lock would be released with the first of them, where a
+        session-scoped advisory lock spans them all.
+        """
+        return ADVISORY_LOCK_KEYS['custom-field-data'], pk
+
+    @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,
+        )
 
-        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.
+    @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).
+        """
+        from extras.jobs import CustomFieldProvisioningJob
+
         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.filter(pk=self.pk).update(status=self.status)
+
+        # 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]
+            )
+        )
+
+    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.
+
+        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.)
+        """
         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 +515,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 +540,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 +555,93 @@ class CustomField(CloningMixin, ExportTemplatesMixin, OwnerMixin, ChangeLoggedMo
                         function='jsonb_set')
                 )
 
+    def delete(self, *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 an update which may run for hours (see CUSTOMFIELD_JOB_TIMEOUT). A field
+        stranded in a pending state by a job which never ran holds no lock, and stays deletable.
+        """
+        if self.status == CustomFieldStatusChoices.STATUS_DELETING:
+            # Already pending deletion; the purge job will remove the row once its data is gone.
+            return 0, {}
+
+        from extras.jobs import CustomFieldPurgeJob
+
+        using = kwargs.get('using') or router.db_for_write(self.__class__, instance=self)
+
+        # Taken outside the transaction below, and so released only once that transaction has ended:
+        # an advisory lock survives a rollback, and releasing it needs a connection in a usable
+        # state, which a transaction aborted by a failed statement is not.
+        with advisory_lock(self.data_lock_key(self.pk), wait=False, using=using) as acquired:
+            if not acquired:
+                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)
+                )
+
+            # Marking the field and dispatching the signals form a single unit of work: a receiver
+            # which raises -- handle_deleted_object() turning a failed deletion protection rule into
+            # AbortRequest, say -- must leave the field exactly as it was. Not every caller provides
+            # a transaction to roll the marking back: ObjectDeleteView deletes in autocommit, which
+            # would otherwise commit the marking and strand the field, unusable and awaiting a purge
+            # job which was never enqueued. Hence a transaction of our own rather than reliance on an
+            # enclosing one.
+            with transaction.atomic(using=using):
+
+                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(*args, **kwargs)
+
+                # Marked before the signals are dispatched, so that handle_cf_deleted() leaves the
+                # stored data to the purge job. Applied via the queryset to avoid emitting a
+                # spurious "updated" change record for a field which has, as far as every consumer
+                # is concerned, just been deleted.
+                self.status = CustomFieldStatusChoices.STATUS_DELETING
+                self.__class__.objects.using(using).filter(pk=self.pk).update(status=self.status)
+
+                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. The purge job takes
+                # this same lock, and waits on it where the deletion enqueued it rather than
+                # housekeeping, so it cannot be lost to the window between that commit and the
+                # release below.
+                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

@@ -4,6 +4,7 @@ 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
@@ -47,11 +48,11 @@ def handle_cf_object_types_changed(instance, action, pk_set, reverse, **kwargs):
     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
+        # Remove the field's stored data from objects to which it no longer applies.
         instance.remove_stale_data(object_types)
 
 
@@ -66,8 +67,12 @@ 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())
 
 
 post_save.connect(handle_cf_renamed, sender=CustomField)

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_customfields.py

@@ -1,30 +1,44 @@
 import datetime
 import json
+import uuid
 from collections import defaultdict
+from contextlib import contextmanager
 from decimal import Decimal
 from unittest.mock import patch
 
 import django_filters
 from django.core.exceptions import ValidationError
-from django.db import connection
+from django.db import DEFAULT_DB_ALIAS, connection, connections
 from django.db.models import QuerySet
-from django.test import override_settings, tag
+from django.db.models.signals import pre_delete
+from django.test import RequestFactory, override_settings, tag
 from django.test.utils import CaptureQueriesContext
 from django.urls import reverse
 from rest_framework import status
 
-from core.models import ObjectChange, ObjectType
+from core.choices import ObjectChangeActionChoices
+from core.jobs import SystemHousekeepingJob
+from core.models import Job, ObjectChange, ObjectType
 from dcim.filtersets import SiteFilterSet
 from dcim.forms import SiteImportForm
 from dcim.models import Manufacturer, Rack, Site
 from dcim.tables import SiteTable
 from extras.choices import *
+from extras.constants import CUSTOMFIELD_JOB_TIMEOUT
 from extras.filters import MissingKeyAwareFilterMixin, missing_key_aware_filter_factory
+from extras.jobs import (
+    CustomFieldProvisioningJob,
+    CustomFieldPurgeJob,
+    provision_custom_field,
+    purge_custom_field,
+)
 from extras.models import CustomField, CustomFieldChoiceSet
 from ipam.models import VLAN
 from netbox.choices import CSVDelimiterChoices, ImportFormatChoices
 from netbox.context import query_cache
+from netbox.context_managers import event_tracking
 from netbox.tables.columns import CustomFieldColumn
+from utilities.exceptions import AbortRequest
 from utilities.filters import MultiValueCharFilter, MultiValueMACAddressFilter
 from utilities.testing import APITestCase, TestCase
 from virtualization.models import VirtualMachine
@@ -688,6 +702,10 @@ class CustomFieldTestCase(TestCase):
         """
         Provisioning, renaming, and removing custom field data is applied in batches. Use a small
         batch size to ensure the data on every object is updated across multiple batches.
+
+        BULK_UPDATE_CHUNK_SIZE doubles as the threshold above which an update is handed to a
+        background job, so overriding it this low also puts provisioning and removal onto the
+        deferred path; the jobs are run here in place of the worker which would ordinarily do so.
         """
         # The existing sites (created in setUpTestData) span multiple batches of size 2
         site_count = Site.objects.count()
@@ -700,12 +718,15 @@ class CustomFieldTestCase(TestCase):
             default='foo'
         )
         cf.object_types.set([self.object_type])
+        self.assertTrue(provision_custom_field(cf.pk))
         self.assertEqual(
             Site.objects.filter(custom_field_data__batched_field='foo').count(),
             site_count
         )
 
-        # Renaming: the key is renamed on every existing object, preserving its value
+        # Renaming: the key is renamed on every existing object, preserving its value. This is
+        # always applied inline, so no job is involved.
+        cf.refresh_from_db()
         cf.name = 'renamed_field'
         cf.save()
         self.assertEqual(
@@ -719,6 +740,7 @@ class CustomFieldTestCase(TestCase):
 
         # Removal: deleting the field strips the key from every existing object
         cf.delete()
+        self.assertTrue(purge_custom_field(cf.pk))
         self.assertEqual(
             Site.objects.filter(custom_field_data__has_key='renamed_field').count(),
             0
@@ -1091,24 +1113,69 @@ class CustomFieldManagerTestCase(TestCase):
         custom_field.object_types.set([object_type])
 
     def test_get_for_model(self):
-        self.assertEqual(CustomField.objects.get_for_model(Site).count(), 1)
-        self.assertEqual(CustomField.objects.get_for_model(VirtualMachine).count(), 0)
+        self.assertEqual(len(CustomField.objects.get_for_model(Site)), 1)
+        self.assertEqual(len(CustomField.objects.get_for_model(VirtualMachine)), 0)
 
     def test_get_for_model_caches_models_with_no_custom_fields(self):
         """
         A model with no custom fields assigned must be served from the request cache like any other.
-        An empty QuerySet is falsy, so testing the cached value for truthiness would treat it as a
-        miss and re-query on every call.
+        An empty list is falsy, so testing the cached value for truthiness would treat it as a miss
+        and re-query on every call.
         """
         token = query_cache.set(defaultdict(dict))
         self.addCleanup(query_cache.reset, token)
 
         # Site has one custom field assigned, VirtualMachine none
         for model in (Site, VirtualMachine):
-            # Prime the cache, iterating so that the QuerySet's own result cache is populated too
-            list(CustomField.objects.get_for_model(model))
+            CustomField.objects.get_for_model(model)  # Prime the cache
+            with self.assertNumQueries(0):
+                CustomField.objects.get_for_model(model)
+
+    def test_get_defaults_for_model_is_cached(self):
+        """
+        Every save of a custom-field-bearing object resolves the model's defaults, so the lookup
+        must be served from the request cache rather than re-queried each time. As above, a model
+        with no defaults caches an empty dict, which must not be mistaken for a miss.
+        """
+        token = query_cache.set(defaultdict(dict))
+        self.addCleanup(query_cache.reset, token)
+
+        # Site has a field with a default, VirtualMachine none
+        for model in (Site, VirtualMachine):
+            CustomField.objects.get_defaults_for_model(model)
             with self.assertNumQueries(0):
-                list(CustomField.objects.get_for_model(model))
+                CustomField.objects.get_defaults_for_model(model)
+
+    def test_get_defaults_for_model_shares_the_field_cache(self):
+        """
+        The two lookups differ only in the statuses they select, so resolving a model's defaults must
+        be served from the fields get_for_model() has already fetched rather than re-querying them.
+        """
+        token = query_cache.set(defaultdict(dict))
+        self.addCleanup(query_cache.reset, token)
+
+        CustomField.objects.get_for_model(Site)  # Prime the field cache
+
+        with self.assertNumQueries(0):
+            self.assertEqual(CustomField.objects.get_defaults_for_model(Site), {'text_field': 'foo'})
+
+    def test_repeated_saves_do_not_requery_custom_fields(self):
+        """
+        A bulk import creates thousands of objects within one request; resolving the defaults afresh
+        for each would add a query per object (see CustomFieldsMixin.save()).
+        """
+        token = query_cache.set(defaultdict(dict))
+        self.addCleanup(query_cache.reset, token)
+
+        Site.objects.create(name='Site 1', slug='site-1')  # Prime the caches
+
+        with CaptureQueriesContext(connection) as ctx:
+            for i in range(2, 5):
+                Site.objects.create(name=f'Site {i}', slug=f'site-{i}')
+
+        custom_field_queries = [q for q in ctx.captured_queries if 'extras_customfield' in q['sql']]
+        self.assertEqual(custom_field_queries, [])
+        self.assertEqual(Site.objects.filter(custom_field_data__text_field='foo').count(), 4)
 
 
 class CustomFieldAPITestCase(APITestCase):
@@ -2711,3 +2778,759 @@ class CustomFieldModelFilterTestCase(TestCase):
             3
         )
         self.assertEqual(self.filterset({'cf_cf12__empty': True}, self.queryset).qs.count(), 1)
+
+
+@contextmanager
+def hold_data_lock(custom_field):
+    """
+    Hold a custom field's data lock on a connection of its own, as a running background job does.
+
+    A separate connection is what makes the lock observable: it is held for the duration of a job,
+    which spans many transactions, so a test cannot take it on the connection it is testing.
+    """
+    lock_key = CustomField.data_lock_key(custom_field.pk)
+    connection = connections.create_connection(DEFAULT_DB_ALIAS)
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute('SELECT pg_try_advisory_lock(%s, %s)', lock_key)
+            if not cursor.fetchone()[0]:
+                raise RuntimeError(f"Failed to acquire the data lock for {custom_field}")
+        yield
+    finally:
+        # Closing the session releases any advisory lock held on it
+        connection.close()
+
+
+@override_settings(BULK_UPDATE_CHUNK_SIZE=1)
+class DeferredCustomFieldDataTestCase(TestCase):
+    """
+    Where too many objects are affected to update within the request, provisioning and purging
+    custom field data is handed to a background job and the field is not live until it completes.
+
+    BULK_UPDATE_CHUNK_SIZE (which doubles as the threshold for deferral) is overridden down so that
+    the two objects below force the deferred path. It cannot be set to zero, which the setting
+    rejects, and which would also empty every batch so that the jobs updated nothing.
+    """
+    @classmethod
+    def setUpTestData(cls):
+        Site.objects.bulk_create([
+            Site(name='Site A', slug='site-a'),
+            Site(name='Site B', slug='site-b'),
+        ])
+        cls.object_type = ObjectType.objects.get_for_model(Site)
+
+    def create_field(self, name='field1', **kwargs):
+        cf = CustomField.objects.create(name=name, type=CustomFieldTypeChoices.TYPE_TEXT, **kwargs)
+        cf.object_types.set([self.object_type])
+        cf.refresh_from_db()
+        return cf
+
+    #
+    # Provisioning
+    #
+
+    def test_provisioning_is_deferred(self):
+        cf = self.create_field(default='foo')
+
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_PROVISIONING)
+        # No object data has been written yet
+        self.assertEqual(Site.objects.filter(custom_field_data__has_key='field1').count(), 0)
+
+    def test_field_is_not_live_while_provisioning(self):
+        cf = self.create_field(default='foo')
+        site = Site.objects.first()
+
+        self.assertNotIn(cf, CustomField.objects.get_for_model(Site))
+        self.assertNotIn('field1', site.cf)
+        self.assertNotIn('field1', {f.name for f in site.get_custom_fields()})
+
+        # It is still reachable where a caller asks for that status, as get_defaults_for_model() does
+        self.assertIn(cf, CustomField.objects.get_for_model(
+            Site, statuses=(CustomFieldStatusChoices.STATUS_PROVISIONING,)
+        ))
+
+    def test_new_objects_receive_default_while_provisioning(self):
+        """
+        A field is provisioned precisely because it carries a default, so an object created while
+        the backfill runs must still receive that default -- the job backfills only what predates
+        the field.
+        """
+        self.create_field(default='foo')
+
+        site = Site.objects.create(name='Site C', slug='site-c')
+
+        site.refresh_from_db()
+        self.assertEqual(site.custom_field_data['field1'], 'foo')
+
+    def test_provisioning_job_backfills_and_activates(self):
+        cf = self.create_field(default='foo')
+
+        self.assertTrue(provision_custom_field(cf.pk))
+
+        cf.refresh_from_db()
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_ACTIVE)
+        self.assertEqual(Site.objects.filter(custom_field_data__field1='foo').count(), 2)
+        self.assertIn(cf, CustomField.objects.get_for_model(Site))
+
+    def test_provisioning_job_commits_each_batch(self):
+        """
+        One transaction spanning the whole backfill would hold a row lock on every object it had
+        rewritten until it finished, for as long as CUSTOMFIELD_JOB_TIMEOUT allows the job to run
+        (see CustomField._update_object_data()).
+        """
+        cf = self.create_field(default='foo')
+
+        with patch.object(CustomField, '_update_object_data') as update:
+            provision_custom_field(cf.pk)
+
+        update.assert_called()
+        for call in update.call_args_list:
+            self.assertTrue(call.kwargs['commit_per_batch'])
+
+    def test_provisioning_job_is_idempotent(self):
+        cf = self.create_field(default='foo')
+        provision_custom_field(cf.pk)
+
+        # A second run finds the field no longer awaiting provisioning and does nothing
+        self.assertFalse(provision_custom_field(cf.pk))
+
+    def test_provisioning_job_overrides_the_default_timeout(self):
+        """
+        The job is enqueued precisely because the work exceeds what a request can absorb, so it must
+        not inherit RQ's default timeout, which is of the same order (see CUSTOMFIELD_JOB_TIMEOUT).
+        """
+        with patch.object(CustomFieldProvisioningJob, 'enqueue') as enqueue:
+            with self.captureOnCommitCallbacks(execute=True):
+                cf = self.create_field(default='foo')
+
+        enqueue.assert_called_once()
+        self.assertEqual(enqueue.call_args.kwargs['job_timeout'], CUSTOMFIELD_JOB_TIMEOUT)
+        self.assertEqual(enqueue.call_args.kwargs['custom_field_pk'], cf.pk)
+
+    def test_provisioning_job_is_enqueued(self):
+        """
+        The Job record itself must be valid: a custom field cannot be assigned to a Job as its
+        object, so the field is identified by primary key instead (see CustomFieldDataJob).
+        """
+        with patch('core.models.jobs.django_rq') as django_rq:
+            with self.captureOnCommitCallbacks(execute=True):
+                cf = self.create_field(default='foo')
+
+        job = Job.objects.get(name__startswith=CustomFieldProvisioningJob.name)
+        self.assertIsNone(job.object_type)
+        self.assertIn(str(cf), job.name)
+        self.assertEqual(
+            django_rq.get_queue.return_value.enqueue.call_args.kwargs['custom_field_pk'], cf.pk
+        )
+
+    def test_provisioning_is_scoped_to_the_new_object_types(self):
+        """
+        Assigning a further object type provisions only that type. The job cannot work this out for
+        itself once the assignment is made, so the types are carried to it.
+        """
+        cf = self.create_field()
+        cf.default = 'foo'
+        cf.save()
+        rack_type = ObjectType.objects.get_for_model(Rack)
+        site = Site.objects.first()
+        Rack.objects.bulk_create([
+            Rack(name='Rack 1', site=site),
+            Rack(name='Rack 2', site=site),
+        ])
+
+        with patch.object(CustomFieldProvisioningJob, 'enqueue') as enqueue:
+            with self.captureOnCommitCallbacks(execute=True):
+                cf.object_types.add(rack_type)
+
+        enqueue.assert_called_once()
+        self.assertEqual(enqueue.call_args.kwargs['object_type_pks'], [rack_type.pk])
+
+    def test_deferral_weighs_only_the_new_object_types(self):
+        """
+        A field already assigned to a large table stays inline when assigned a small one: the tables
+        provisioned previously are not rewritten, so their size is beside the point.
+        """
+        cf = self.create_field()
+        cf.default = 'foo'
+        cf.save()
+
+        # No racks exist, so there is nothing to defer even though the two sites exceed the limit
+        cf.object_types.add(ObjectType.objects.get_for_model(Rack))
+
+        cf.refresh_from_db()
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_ACTIVE)
+
+    def test_housekeeping_provisions_every_assigned_type(self):
+        """
+        The backstop has no record of which types a deferred job was to provision, so it falls back
+        to all of them. It must still not disturb the values already stored.
+        """
+        cf = self.create_field(default='foo')
+        Site.objects.update(custom_field_data={'field1': 'bar'})
+
+        self.assertTrue(provision_custom_field(cf.pk))
+
+        cf.refresh_from_db()
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_ACTIVE)
+        self.assertEqual(Site.objects.filter(custom_field_data__field1='bar').count(), 2)
+
+    def test_field_without_default_is_not_deferred(self):
+        """
+        A field with no default has nothing to provision, so it goes live immediately regardless of
+        how many objects it applies to.
+        """
+        cf = self.create_field()
+
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_ACTIVE)
+
+    def test_field_without_default_enqueues_nothing(self):
+        """
+        The decision rests with provision_data() rather than its caller, so a field with no default
+        must not reach the point of sizing its object types, let alone of handing a job the no-op of
+        writing a null to each of them.
+        """
+        cf = self.create_field()
+
+        with (
+            patch.object(CustomField, '_exceeds_inline_limit') as exceeds_limit,
+            patch.object(CustomFieldProvisioningJob, 'enqueue') as enqueue,
+        ):
+            with self.captureOnCommitCallbacks(execute=True):
+                cf.provision_data([self.object_type])
+
+        exceeds_limit.assert_not_called()
+        enqueue.assert_not_called()
+
+    #
+    # Deletion
+    #
+
+    def test_deletion_is_deferred(self):
+        cf = self.create_field()
+        Site.objects.update(custom_field_data={'field1': 'foo'})
+
+        cf.delete()
+
+        cf = CustomField.objects.get(pk=cf.pk)
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_DELETING)
+        # The stored data is left for the purge job
+        self.assertEqual(Site.objects.filter(custom_field_data__has_key='field1').count(), 2)
+
+    def test_deletion_is_deferred_even_without_stored_data(self):
+        """
+        The deferral decision weighs every row of the assigned types, not just those which hold a
+        value, so a field holding no data on an over-limit table is still deferred. Deliberate: the
+        probe cannot count the rows holding a key without a sequential scan (see
+        _exceeds_inline_limit()), and the purge job it hands off to has nothing to do.
+        """
+        cf = self.create_field()
+        self.assertEqual(Site.objects.filter(custom_field_data__has_key='field1').count(), 0)
+
+        cf.delete()
+
+        cf = CustomField.objects.get(pk=cf.pk)
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_DELETING)
+        self.assertTrue(purge_custom_field(cf.pk))
+        self.assertFalse(CustomField.objects.filter(pk=cf.pk).exists())
+
+    def test_field_is_not_live_while_deleting(self):
+        cf = self.create_field()
+        Site.objects.update(custom_field_data={'field1': 'foo'})
+        cf.delete()
+
+        site = Site.objects.first()
+        self.assertNotIn(cf, CustomField.objects.get_for_model(Site))
+        self.assertNotIn('field1', site.cf)
+        self.assertNotIn('field1', {f.name for f in site.get_custom_fields()})
+
+    def test_purge_job_removes_data_and_field(self):
+        cf = self.create_field()
+        Site.objects.update(custom_field_data={'field1': 'foo'})
+        cf.delete()
+
+        self.assertTrue(purge_custom_field(cf.pk))
+
+        self.assertFalse(CustomField.objects.filter(pk=cf.pk).exists())
+        self.assertEqual(Site.objects.filter(custom_field_data__has_key='field1').count(), 0)
+
+    def test_purge_job_commits_each_batch(self):
+        cf = self.create_field()
+        Site.objects.update(custom_field_data={'field1': 'foo'})
+        cf.delete()
+
+        with patch.object(CustomField, '_update_object_data') as update:
+            purge_custom_field(cf.pk)
+
+        update.assert_called()
+        for call in update.call_args_list:
+            self.assertTrue(call.kwargs['commit_per_batch'])
+
+    def test_purge_job_is_idempotent(self):
+        cf = self.create_field()
+        cf.delete()
+        purge_custom_field(cf.pk)
+
+        # A second run finds the field already gone and does nothing
+        self.assertFalse(purge_custom_field(cf.pk))
+
+    def test_deleting_twice_is_a_noop(self):
+        cf = self.create_field()
+        cf.delete()
+
+        cf.delete()
+
+        self.assertTrue(CustomField.objects.filter(pk=cf.pk).exists())
+
+    def test_aborted_deletion_leaves_the_field_intact(self):
+        """
+        A receiver rejecting the deletion (e.g. handle_deleted_object() raising AbortRequest for a
+        failed protection rule) must leave the field live, rather than marked for a purge which will
+        never be enqueued -- and which housekeeping would later complete, destroying the very data
+        the rule protected.
+        """
+        cf = self.create_field()
+        Site.objects.update(custom_field_data={'field1': 'foo'})
+
+        def reject(sender, instance, **kwargs):
+            raise AbortRequest("Deletion is prevented by a protection rule")
+
+        pre_delete.connect(reject, sender=CustomField)
+        try:
+            with self.assertRaises(AbortRequest):
+                cf.delete()
+        finally:
+            pre_delete.disconnect(reject, sender=CustomField)
+
+        cf = CustomField.objects.get(pk=cf.pk)
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_ACTIVE)
+        self.assertIn(cf, CustomField.objects.get_for_model(Site))
+        self.assertEqual(Site.objects.filter(custom_field_data__field1='foo').count(), 2)
+
+    def test_purge_job_overrides_the_default_timeout(self):
+        cf = self.create_field()
+        Site.objects.update(custom_field_data={'field1': 'foo'})
+
+        with patch.object(CustomFieldPurgeJob, 'enqueue') as enqueue:
+            with self.captureOnCommitCallbacks(execute=True):
+                cf.delete()
+
+        enqueue.assert_called_once()
+        self.assertEqual(enqueue.call_args.kwargs['job_timeout'], CUSTOMFIELD_JOB_TIMEOUT)
+        self.assertEqual(enqueue.call_args.kwargs['custom_field_pk'], cf.pk)
+
+    def test_purge_job_is_enqueued(self):
+        cf = self.create_field()
+        Site.objects.update(custom_field_data={'field1': 'foo'})
+
+        with patch('core.models.jobs.django_rq') as django_rq:
+            with self.captureOnCommitCallbacks(execute=True):
+                cf.delete()
+
+        job = Job.objects.get(name__startswith=CustomFieldPurgeJob.name)
+        self.assertIsNone(job.object_type)
+        self.assertIn(str(cf), job.name)
+        self.assertEqual(
+            django_rq.get_queue.return_value.enqueue.call_args.kwargs['custom_field_pk'], cf.pk
+        )
+
+    def test_deletion_records_a_change(self):
+        """
+        The change log must report the deletion where the user performed it, rather than when the
+        row is eventually removed in a worker (where there is no request to attribute it to).
+        """
+        cf = self.create_field()
+
+        request = RequestFactory().get('/')
+        request.id = uuid.uuid4()
+        request.user = self.user
+        with event_tracking(request):
+            cf.delete()
+
+        self.assertTrue(
+            ObjectChange.objects.filter(
+                changed_object_type=ObjectType.objects.get_for_model(CustomField),
+                changed_object_id=cf.pk,
+                action=ObjectChangeActionChoices.ACTION_DELETE,
+            ).exists()
+        )
+
+    def test_deletion_is_scoped_to_the_write_database(self):
+        """
+        The commit hook must be registered against the connection the marking was written on, or the
+        purge job can be enqueued before -- or without -- the field being durably marked.
+        """
+        cf = self.create_field()
+
+        with patch('extras.models.customfields.transaction.on_commit') as on_commit:
+            cf.delete()
+
+        # transaction.on_commit is patched on the shared module, so hooks registered by unrelated
+        # machinery during the delete (deferred search indexing, for one) are captured here too.
+        # Select the hook which enqueues the purge job rather than assuming it is the only one.
+        calls = [
+            call for call in on_commit.call_args_list
+            if 'CustomField.delete' in getattr(call.args[0], '__qualname__', '')
+        ]
+        self.assertEqual(len(calls), 1)
+        self.assertEqual(calls[0].kwargs['using'], DEFAULT_DB_ALIAS)
+
+    #
+    # Name reservation
+    #
+
+    def test_name_is_reserved_while_deleting(self):
+        """
+        A field pending deletion holds its name, so that a new field cannot inherit the values still
+        stored against it.
+        """
+        cf = self.create_field()
+        Site.objects.update(custom_field_data={'field1': 'foo'})
+        cf.delete()
+
+        replacement = CustomField(name='field1', type=CustomFieldTypeChoices.TYPE_TEXT)
+        with self.assertRaises(ValidationError):
+            replacement.full_clean()
+
+    def test_rename_onto_reserved_name_is_rejected(self):
+        cf = self.create_field()
+        Site.objects.update(custom_field_data={'field1': 'foo'})
+        cf.delete()
+        other = self.create_field(name='field2')
+
+        other.name = 'field1'
+        with self.assertRaises(ValidationError):
+            other.full_clean()
+
+    def test_name_is_released_once_purged(self):
+        cf = self.create_field()
+        cf.delete()
+        purge_custom_field(cf.pk)
+
+        replacement = CustomField(name='field1', type=CustomFieldTypeChoices.TYPE_TEXT)
+        replacement.full_clean()  # Should not raise
+
+    #
+    # Modification and deletion guards
+    #
+
+    def test_pending_field_cannot_be_modified(self):
+        cf = self.create_field(default='foo')
+
+        cf.label = 'Changed'
+        with self.assertRaises(ValidationError):
+            cf.full_clean()
+
+    def test_deletion_claims_the_data_lock_without_waiting(self):
+        """
+        A job holds the field's data lock for the duration of its bulk update, so a deletion which
+        waited on it would occupy a worker for as long as the job ran (see CUSTOMFIELD_JOB_TIMEOUT).
+        """
+        cf = self.create_field()
+
+        with CaptureQueriesContext(connection) as queries:
+            cf.delete()
+
+        self.assertTrue(
+            any('pg_try_advisory_lock' in query['sql'] for query in queries),
+            "Deletion did not claim the field's data lock without waiting"
+        )
+
+    def test_deletion_is_refused_while_a_job_holds_the_data_lock(self):
+        """
+        Failing to take the lock aborts the deletion cleanly, rather than surfacing a database error,
+        and must leave the field exactly as it was.
+        """
+        cf = self.create_field()
+
+        with hold_data_lock(cf):
+            with self.assertRaises(AbortRequest):
+                cf.delete()
+
+            cf.refresh_from_db()
+            self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_ACTIVE)
+
+        cf.delete()  # Released: the deletion now proceeds
+
+        cf.refresh_from_db()
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_DELETING)
+
+    def test_job_skips_a_field_locked_by_another_job(self):
+        """
+        The backstop must not block a worker on a field the responsible job is still working through,
+        which may be hours from completing.
+        """
+        cf = self.create_field(default='foo')
+
+        with hold_data_lock(cf):
+            self.assertFalse(provision_custom_field(cf.pk, skip_locked=True))
+
+        cf.refresh_from_db()
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_PROVISIONING)
+        self.assertEqual(Site.objects.filter(custom_field_data__has_key='field1').count(), 0)
+
+    def test_stranded_field_can_be_deleted(self):
+        """
+        The refusal is on the lock, not on the status: a field left mid-provisioning by a job which
+        never ran holds no lock, and must remain deletable without waiting for housekeeping.
+        """
+        cf = self.create_field(default='foo')
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_PROVISIONING)
+
+        cf.delete()
+
+        cf.refresh_from_db()
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_DELETING)
+
+    #
+    # Housekeeping backstop
+    #
+
+    def test_housekeeping_enqueues_pending_fields(self):
+        """
+        A field whose job never ran is picked up by the daily housekeeping job, so that it cannot
+        remain offline (or holding its name) indefinitely. The work is handed back to a dedicated
+        job rather than performed inline, where it would be subject to housekeeping's own timeout.
+        """
+        provisioning = self.create_field(name='field1', default='foo')
+        deleting = self.create_field(name='field2')
+        deleting.delete()
+
+        with (
+            patch.object(CustomFieldProvisioningJob, 'enqueue') as provision,
+            patch.object(CustomFieldPurgeJob, 'enqueue') as purge,
+        ):
+            SystemHousekeepingJob(Job()).finalize_custom_fields()
+
+        for enqueue, custom_field in ((provision, provisioning), (purge, deleting)):
+            enqueue.assert_called_once()
+            self.assertEqual(enqueue.call_args.kwargs['custom_field_pk'], custom_field.pk)
+            self.assertEqual(enqueue.call_args.kwargs['job_timeout'], CUSTOMFIELD_JOB_TIMEOUT)
+            self.assertTrue(enqueue.call_args.kwargs['skip_locked'])
+
+    def test_housekeeping_leaves_active_fields_alone(self):
+        self.create_field(name='field1')
+
+        with (
+            patch.object(CustomFieldProvisioningJob, 'enqueue') as provision,
+            patch.object(CustomFieldPurgeJob, 'enqueue') as purge,
+        ):
+            SystemHousekeepingJob(Job()).finalize_custom_fields()
+
+        provision.assert_not_called()
+        purge.assert_not_called()
+
+    def test_job_forwards_skip_locked(self):
+        """
+        The job enqueued by housekeeping must not wait on a field the responsible job still holds:
+        that job may be hours from completing, and blocking here occupies a worker for as long. So
+        the flag has to reach the lock, rather than being swallowed by run().
+        """
+        cf = self.create_field(default='foo')
+
+        with patch('extras.jobs.provision_custom_field') as provision:
+            CustomFieldProvisioningJob(Job()).run(custom_field_pk=cf.pk, skip_locked=True)
+
+        provision.assert_called_once_with(cf.pk, None, skip_locked=True)
+
+    def test_housekeeping_completes_pending_fields(self):
+        """
+        End to end: the jobs housekeeping enqueues bring a stranded field to a resolved state.
+        """
+        provisioning = self.create_field(name='field1', default='foo')
+        deleting = self.create_field(name='field2')
+        deleting.delete()
+
+        with patch('core.models.jobs.django_rq'):
+            SystemHousekeepingJob(Job()).finalize_custom_fields()
+        for custom_field, job_class in (
+            (provisioning, CustomFieldProvisioningJob),
+            (deleting, CustomFieldPurgeJob),
+        ):
+            job_class(Job()).run(custom_field_pk=custom_field.pk, skip_locked=True)
+
+        provisioning.refresh_from_db()
+        self.assertEqual(provisioning.status, CustomFieldStatusChoices.STATUS_ACTIVE)
+        self.assertEqual(Site.objects.filter(custom_field_data__field1='foo').count(), 2)
+        self.assertFalse(CustomField.objects.filter(pk=deleting.pk).exists())
+
+
+class InlineCustomFieldDataTestCase(TestCase):
+    """
+    Where few enough objects are affected, provisioning and purging remain synchronous: the field is
+    live (or gone) as soon as the request completes, with no background job involved.
+    """
+    @classmethod
+    def setUpTestData(cls):
+        Site.objects.create(name='Site A', slug='site-a')
+        cls.object_type = ObjectType.objects.get_for_model(Site)
+
+    def test_provisioning_is_inline(self):
+        cf = CustomField.objects.create(
+            name='field1', type=CustomFieldTypeChoices.TYPE_TEXT, default='foo'
+        )
+        cf.object_types.set([self.object_type])
+
+        cf.refresh_from_db()
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_ACTIVE)
+        self.assertEqual(Site.objects.filter(custom_field_data__field1='foo').count(), 1)
+
+    def test_inline_provisioning_is_atomic(self):
+        """
+        The request path answers to an enclosing transaction, which owns the commit: committing each
+        batch there would silently do nothing, and a failure part-way must leave nothing behind.
+        """
+        with patch.object(CustomField, '_update_object_data') as update:
+            cf = CustomField.objects.create(
+                name='field1', type=CustomFieldTypeChoices.TYPE_TEXT, default='foo'
+            )
+            cf.object_types.set([self.object_type])
+
+        update.assert_called()
+        for call in update.call_args_list:
+            self.assertFalse(call.kwargs['commit_per_batch'])
+
+    def test_default_added_later_is_not_backfilled(self):
+        """
+        A default added to a field which already exists is not backfilled, and assigning a further
+        object type must not backfill it either: only the newly assigned type is provisioned. The
+        sites below would otherwise acquire a value they were documented never to receive.
+        """
+        cf = CustomField.objects.create(name='field1', type=CustomFieldTypeChoices.TYPE_TEXT)
+        cf.object_types.set([self.object_type])
+        self.assertEqual(Site.objects.first().custom_field_data, {})
+
+        cf.default = 'foo'
+        cf.save()
+        self.assertEqual(Site.objects.first().custom_field_data, {})
+
+        rack = Rack.objects.create(name='Rack 1', site=Site.objects.first())
+        cf.object_types.add(ObjectType.objects.get_for_model(Rack))
+
+        # The newly assigned type is provisioned; the one assigned before the default is not
+        rack.refresh_from_db()
+        self.assertEqual(rack.custom_field_data['field1'], 'foo')
+        self.assertEqual(Site.objects.first().custom_field_data, {})
+
+    def test_provisioning_preserves_existing_values(self):
+        """
+        Values stored against a type assigned previously must survive a further assignment.
+        """
+        cf = CustomField.objects.create(
+            name='field1', type=CustomFieldTypeChoices.TYPE_TEXT, default='foo'
+        )
+        cf.object_types.set([self.object_type])
+        Site.objects.update(custom_field_data={'field1': 'bar'})
+
+        cf.object_types.add(ObjectType.objects.get_for_model(Rack))
+
+        self.assertEqual(Site.objects.first().custom_field_data['field1'], 'bar')
+
+    def test_provisioning_preserves_cleared_values(self):
+        """
+        A cleared value is stored as a JSON null rather than an absent key, and must survive
+        reprovisioning just as a set value does.
+        """
+        cf = CustomField.objects.create(
+            name='field1', type=CustomFieldTypeChoices.TYPE_TEXT, default='foo'
+        )
+        cf.object_types.set([self.object_type])
+        Site.objects.update(custom_field_data={'field1': None})
+
+        cf.object_types.add(ObjectType.objects.get_for_model(Rack))
+
+        self.assertIsNone(Site.objects.first().custom_field_data['field1'])
+
+    def test_deletion_is_inline(self):
+        cf = CustomField.objects.create(name='field1', type=CustomFieldTypeChoices.TYPE_TEXT)
+        cf.object_types.set([self.object_type])
+        Site.objects.update(custom_field_data={'field1': 'foo'})
+
+        cf.delete()
+
+        self.assertFalse(CustomField.objects.filter(pk=cf.pk).exists())
+        self.assertEqual(Site.objects.filter(custom_field_data__has_key='field1').count(), 0)
+
+
+@override_settings(BULK_UPDATE_CHUNK_SIZE=None)
+class UnchunkedCustomFieldDataTestCase(TestCase):
+    """
+    Setting BULK_UPDATE_CHUNK_SIZE to None disables chunking, so a bulk update is issued as a single
+    unbounded statement. There is then no batch size for the deferral threshold to test against, and
+    an unbounded JSONB rewrite is exactly what must not run inside a request -- so any affected
+    object sends the work to a background job, which issues that one statement under a timeout
+    generous enough to survive it.
+    """
+    @classmethod
+    def setUpTestData(cls):
+        Site.objects.create(name='Site A', slug='site-a')
+        cls.object_type = ObjectType.objects.get_for_model(Site)
+
+    @staticmethod
+    def _count_updates(queries, model):
+        """
+        Count the UPDATE statements issued against the given model's table, ignoring those the job
+        makes to the custom field row itself (marking it active).
+        """
+        table = model._meta.db_table
+        return len([
+            q for q in queries
+            if q['sql'].strip().upper().startswith('UPDATE') and table in q['sql']
+        ])
+
+    def test_provisioning_is_deferred(self):
+        """
+        A single object is enough: with chunking disabled there is no bound on the statement the
+        request would otherwise issue.
+        """
+        cf = CustomField.objects.create(
+            name='field1', type=CustomFieldTypeChoices.TYPE_TEXT, default='foo'
+        )
+        cf.object_types.set([self.object_type])
+
+        cf.refresh_from_db()
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_PROVISIONING)
+        self.assertEqual(Site.objects.filter(custom_field_data__has_key='field1').count(), 0)
+
+    def test_provisioning_job_backfills_in_a_single_statement(self):
+        cf = CustomField.objects.create(
+            name='field1', type=CustomFieldTypeChoices.TYPE_TEXT, default='foo'
+        )
+        cf.object_types.set([self.object_type])
+
+        with CaptureQueriesContext(connection) as queries:
+            self.assertTrue(provision_custom_field(cf.pk))
+
+        # One statement covers the table, rather than one per batch
+        self.assertEqual(self._count_updates(queries.captured_queries, Site), 1)
+
+        cf.refresh_from_db()
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_ACTIVE)
+        self.assertEqual(Site.objects.filter(custom_field_data__field1='foo').count(), 1)
+
+    def test_field_affecting_no_objects_stays_inline(self):
+        """
+        A limit of zero still leaves the probe testing for a single row, so a field which rewrites
+        nothing goes live in the request rather than waiting on a job with no work to do.
+        """
+        cf = CustomField.objects.create(
+            name='field1', type=CustomFieldTypeChoices.TYPE_TEXT, default='foo'
+        )
+
+        # No racks exist, so there is nothing to rewrite
+        cf.object_types.set([ObjectType.objects.get_for_model(Rack)])
+
+        cf.refresh_from_db()
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_ACTIVE)
+
+    def test_deletion_is_deferred(self):
+        cf = CustomField.objects.create(name='field1', type=CustomFieldTypeChoices.TYPE_TEXT)
+        cf.object_types.set([self.object_type])
+        Site.objects.update(custom_field_data={'field1': 'foo'})
+
+        cf.delete()
+
+        cf.refresh_from_db()
+        self.assertEqual(cf.status, CustomFieldStatusChoices.STATUS_DELETING)
+        self.assertTrue(purge_custom_field(cf.pk))
+        self.assertFalse(CustomField.objects.filter(pk=cf.pk).exists())
+        self.assertEqual(Site.objects.filter(custom_field_data__has_key='field1').count(), 0)

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_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

@@ -227,7 +227,7 @@ class CustomFieldsMixin(models.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)
@@ -337,10 +338,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,12 @@ 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.
     """
     if chunk_size is None:
         chunk_size = settings.BULK_UPDATE_CHUNK_SIZE
@@ -48,7 +56,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: