Closes #23259: Coalesce search cache updates instead of enqueuing a job per change (#23378)

* fix(search): Coalesce deferred index updates into a single drain

* fix(search): Drain coalesced updates and add a safety-net job

* fix(search): Register the search-cache safety-net job at startup

* test(search): Cover search-cache coalescing and recovery

* docs(search): Document deferred search indexing and recovery

* test(search): Serialize search-cache tests against RQ queue suites

The search-cache tests use the real tasks Redis (DB 0) that RQQueueTestMixin
also flushes via flushdb(). Under the parallel runner those classes run
concurrently and a worker's flush wiped the coalescing keys mid-test, causing
the coalescing and stranded-staging tests to fail non-deterministically.
Applying RQQueueTestMixin serializes them against the other Redis-backed
suites and supplies the DB-0 cleanup, so the ad-hoc _clear_search_redis helper
is no longer needed.

* fix(docs): Bump mkdocstrings to 1.0.7 for pymdown-extensions 12.2

mkdocstrings 1.0.7 adds support for pymdown-extensions 12.2, which changed
Highlight.__init__ to require the md argument. Without the bump the docs build
crashes on a fresh install (which resolves pymdown-extensions 12.x) with
'Highlight.__init__() missing 1 required positional argument: md'.

* fix(search): Serialize deferred cache applies and cap retries

Hold a PostgreSQL advisory lock across the whole apply-and-cleanup
lifecycle of a staged batch, so the safety net cannot re-apply a batch
whose Redis lease expired while its applier is still running (which
duplicated CachedValue rows under READ COMMITTED). The legacy
SearchCacheJob back-compat path takes the same lock (wait=True).

Cap persistent apply failures at MAX_APPLY_ATTEMPTS and drop the batch
with an error-level log, instead of retrying forever.

docs/features/search.md

@@ -8,6 +8,14 @@ When entering a search query, the user can choose a specific lookup type: exact
 
 Custom fields defined by NetBox administrators are also included in search results if configured with a search weight. Additionally, NetBox plugins can register their own custom models for inclusion alongside core models.
 
+### Deferred indexing
+
+To avoid enqueuing a separate background job for every change, search index updates are coalesced: changes committed over a short interval are buffered and applied together by a single background job. This settles quickly (on the order of a couple of seconds) when workers are running. A recurring safety-net job also recovers any pending updates whose drain was lost (for example, if a worker was stopped mid-run), so indexing is eventually consistent even after a worker restart. Because a background drain can run longer than its Redis lease, overlapping applies are prevented by a database-level lock, so an object's index entries are never duplicated.
+
+The pending buffer is stored in the tasks Redis database. If that Redis data is lost, some recent index updates may be discarded; run `manage.py reindex` to rebuild the search index from the database.
+
+If a buffered batch repeatedly fails to apply (for example, due to a persistent database error), it is retried by the safety-net job a limited number of times and then discarded, with the failure recorded at error level in the system log. The objects in a discarded batch will not be re-indexed until they are saved again; run `manage.py reindex` to rebuild the search index from the database.
+
 !!! note
     NetBox does not index any static choice field's (including custom fields of type "Selection" or "Multiple selection").
 

netbox/core/apps.py

@@ -36,6 +36,12 @@ class CoreConfig(AppConfig):
         )
         from netbox import context_managers  # noqa: F401
         from netbox.models.features import register_models
+
+        # Importing netbox.search.jobs registers its system job(s) (e.g. the
+        # search-cache flush safety net) in registry['system_jobs'] before the
+        # RQ worker reads that registry at startup. The module is otherwise only
+        # imported lazily, so this import is load-bearing, not cosmetic.
+        from netbox.search import jobs as search_jobs  # noqa: F401
         from netbox.search import signals as search_signals  # noqa: F401
 
         from . import data_backends, events, search  # noqa: F401

netbox/netbox/search/deferred.py

@@ -1,7 +1,13 @@
 import logging
+import time
+import uuid
+from datetime import timedelta
 
 from django.db import DEFAULT_DB_ALIAS, connections, transaction
-from redis.exceptions import RedisError
+from django.utils import timezone
+from django_pg_utils import advisory_lock
+from django_rq.queues import get_connection
+from redis.exceptions import RedisError, ResponseError
 
 from netbox.constants import RQ_QUEUE_DEFAULT
 from netbox.search.backends import search_backend
@@ -17,6 +23,29 @@ logger = logging.getLogger(__name__)
 OP_CACHE = 'cache'
 OP_REMOVE = 'remove'
 
+# Cross-transaction coalescing buffer. Committed batches accumulate in a
+# per-alias Redis hash on the tasks connection and are drained by a single
+# scheduled SearchCacheJob, so N committed transactions produce roughly one job
+# instead of N (the regression this replaces).
+FLUSH_INTERVAL = 2  # Seconds to coalesce before draining.
+LOCK_TTL = 60       # Seconds; scheduling throttle lease (one drain scheduled per alias).
+APPLY_LEASE = 60    # Seconds; per-batch processing lease (one applier per batch).
+
+# A persistently failing batch is retried by the safety net until it has failed
+# this many times, then dropped with an error-level log; the affected objects
+# stay unindexed until re-saved or `manage.py reindex` is run. The cap stops one
+# poisoned batch from blocking indexing of unrelated objects forever.
+MAX_APPLY_ATTEMPTS = 5
+
+PENDING_PREFIX = 'netbox:search:pending:'
+LOCK_PREFIX = 'netbox:search:flush-lock:'
+STAGING_PREFIX = 'netbox:search:draining:'
+LEASE_PREFIX = 'netbox:search:draining-lease:'
+ATTEMPTS_PREFIX = 'netbox:search:apply-attempts:'
+# Bounds growth of a batch's failure counter if its staging key disappears
+# through another path (e.g. a worker that crashed before writing the counter).
+ATTEMPTS_TTL = 3600
+
 # Attributes used to tag a flush callback so we can recognize our own callbacks
 # among those registered on a connection and reach the batch they will flush.
 _FLUSH_ALIAS_ATTR = '_netbox_search_flush_alias'
@@ -140,9 +169,11 @@ def _flush(batch, using):
     time it runs the originating write has already committed, so it must never
     propagate an error back to the caller and turn a successful save into a 500.
 
-    The inline fallback is safe even during a broker outage: the search index
-    lives in PostgreSQL (the extras_cachedvalue table), so a Redis outage only
-    prevents backgrounding, not indexing itself.
+    When a worker is available the batch is merged into the shared Redis buffer
+    and a single drain is scheduled (see _buffer_and_schedule); committed batches
+    from many transactions thus coalesce into far fewer jobs than the
+    one-job-per-transaction it replaces. Otherwise it falls back to inline
+    indexing, a PostgreSQL write with no Redis dependency.
 
     If the broker fails mid-enqueue (after the probe succeeds), Job.enqueue() has
     already saved a Job row before the Redis dispatch raised, so the fallback can
@@ -160,14 +191,274 @@ def _flush(batch, using):
         groups.setdefault(object_type_id, []).append(pk)
 
     try:
-        # Both the worker-availability check and the job enqueue talk to Redis,
-        # and a worker can die between the two. Treat any Redis failure across the
-        # whole dispatch as "no worker available" and fall back to inline
-        # indexing (a PostgreSQL write that does not depend on Redis).
-        if any_workers_for_queue(RQ_QUEUE_DEFAULT):
-            SearchCacheJob.enqueue(using=using, cache_groups=cache_groups, remove_groups=remove_groups)
-            return
+        has_worker = any_workers_for_queue(RQ_QUEUE_DEFAULT)
     except RedisError:
-        logger.warning("Search cache: broker unavailable; indexing inline", exc_info=True)
+        has_worker = False
+
+    if has_worker:
+        try:
+            _buffer_and_schedule(batch, using)
+            return
+        except RedisError:
+            # A worker can die (or the broker can drop) between the availability
+            # probe and the buffer write. Fall back to inline indexing. Any
+            # pending entry already written is deliberately left in place: it is
+            # a recoverable duplicate, and re-applying it is safe because
+            # _apply_deferred_updates is idempotent. Deleting it here could
+            # discard a concurrent writer's update, which is worse.
+            logger.warning("Search cache: broker unavailable; indexing inline", exc_info=True)
 
     search_backend._apply_deferred_updates(using=using, cache_groups=cache_groups, remove_groups=remove_groups)
+
+
+def _buffer_and_schedule(batch, using):
+    """
+    Merge a committed batch into the shared pending hash and ensure exactly one
+    drain is scheduled for the alias.
+    """
+    conn = _connection()
+    conn.hset(
+        _pending_key(using),
+        mapping={_field(object_type_id, pk): op for (object_type_id, pk), op in batch.items()},
+    )
+    # Only the flush that wins the throttle lock schedules a drain; the rest just
+    # merge into the buffer. HSET overwrites a field, so a later op for the same
+    # object wins -- the correct cross-batch semantics (a later save supersedes
+    # an earlier delete).
+    if conn.set(_lock_key(using), '1', nx=True, ex=LOCK_TTL):
+        try:
+            _schedule(using)
+        except Exception:
+            # Nothing will run the scheduled drain; release the lock so the
+            # safety net (or the next flush) can pick the buffer up promptly.
+            conn.delete(_lock_key(using))
+            raise
+
+
+def _schedule(using):
+    SearchCacheJob.enqueue(
+        using=using,
+        schedule_at=timezone.now() + timedelta(seconds=FLUSH_INTERVAL),
+    )
+
+
+def drain_and_apply(using, log=None):
+    """
+    Claim and apply the pending batch for a single database alias.
+
+    Called by SearchCacheJob after its scheduled delay. The pending hash is moved
+    atomically to a durable staging key (see _claim) and applied there, so a
+    crash before the work completes leaves the batch recoverable.
+    """
+    staging = _claim(using)
+    if staging is not None:
+        _apply_staging(using, staging, log or logger)
+
+
+def _claim(using):
+    """
+    Atomically take ownership of the pending hash for an alias.
+
+    RENAME is the claim: exactly one drainer can win it. The resulting staging
+    key is immutable (producers only ever write to the pending key), so no update
+    buffered while the batch is applied can be lost or overwritten.
+
+    Returns the staging key, or None if there was nothing pending.
+    """
+    conn = _connection()
+    staging = f'{STAGING_PREFIX}{using}:{int(time.time())}:{uuid.uuid4().hex}'
+    try:
+        conn.rename(_pending_key(using), staging)
+    except ResponseError:
+        return None
+    # Release the throttle once the batch is claimed, so updates arriving during
+    # the (potentially slow) apply can schedule a fresh drain rather than being
+    # stranded until the safety net runs.
+    conn.delete(_lock_key(using))
+    return staging
+
+
+def _apply_staging(using, staging, log):
+    """
+    Apply a claimed staging batch, deleting it only once application succeeds.
+
+    The per-batch Redis lease is only a cheap pre-filter: it is bounded by
+    APPLY_LEASE while the applier can run for up to the job timeout, so the lease
+    alone cannot guarantee exclusive execution (the safety net can take a fresh
+    lease and start applying the same batch while the original is still running).
+    Actual exclusion is provided by a PostgreSQL session-level advisory lock held
+    for the whole apply-and-cleanup lifecycle. `_apply_deferred_updates` is
+    idempotent, so any retry after the lock is released converges.
+
+    The lease is never deleted by its owner (it only expires), so a slow or
+    crashed drainer can never delete a lease a successor has taken over.
+    """
+    conn = _connection()
+    batch_id = staging.rsplit(':', 1)[1]
+
+    if not conn.set(_lease_key(batch_id), '1', nx=True, ex=APPLY_LEASE):
+        # Another drainer holds the lease; let it finish.
+        return
+
+    data = conn.hgetall(staging)
+    if not data:
+        conn.delete(staging)
+        return
+
+    _warn_if_stale(staging, log)
+
+    cache_groups = {}
+    remove_groups = {}
+    for raw_field, raw_op in data.items():
+        field = raw_field.decode() if isinstance(raw_field, bytes) else raw_field
+        op = raw_op.decode() if isinstance(raw_op, bytes) else raw_op
+        object_type_id, pk = field.split(':', 1)
+        groups = remove_groups if op == OP_REMOVE else cache_groups
+        groups.setdefault(int(object_type_id), []).append(int(pk))
+
+    # The advisory lock must span the apply *and* the staging deletion below,
+    # otherwise a second applier could acquire it after the apply but before the
+    # delete and apply the same batch again, leaving duplicate rows.
+    db = using or DEFAULT_DB_ALIAS
+    with advisory_lock(_apply_lock_key(db), wait=False, using=db) as acquired:
+        if not acquired:
+            # Another applier holds the advisory lock: its Redis lease may have
+            # expired while it is still running. Do not apply concurrently (that
+            # would duplicate rows). It deletes staging on success; if it is
+            # killed, PostgreSQL releases the lock when its session ends and the
+            # safety net retries the retained batch. Contention is not an apply
+            # failure, so it is not counted against MAX_APPLY_ATTEMPTS.
+            return
+
+        try:
+            search_backend._apply_deferred_updates(
+                using=using,
+                cache_groups=cache_groups,
+                remove_groups=remove_groups,
+                log=log,
+            )
+        except Exception:
+            if _record_apply_failure(conn, batch_id) >= MAX_APPLY_ATTEMPTS:
+                # Retrying a persistently failing batch forever would block
+                # indexing of unrelated objects and grow staging without bound.
+                # Drop it with an error-level log; the batch's objects stay
+                # unindexed until re-saved or `manage.py reindex` is run.
+                log.error(
+                    f"Search cache: dropping deferred batch {staging} after "
+                    f"{MAX_APPLY_ATTEMPTS} failed apply attempts; run `manage.py reindex` "
+                    f"to rebuild the search index",
+                    exc_info=True,
+                )
+                conn.delete(staging)
+                conn.delete(_attempts_key(batch_id))
+                return
+            raise
+
+        # Retained until this returns without raising: a crash or database error
+        # leaves the staging key for the safety net to retry.
+        conn.delete(staging)
+        conn.delete(_attempts_key(batch_id))
+
+    # Updates that arrived while this batch was being applied land in a fresh
+    # pending hash; schedule one more drain if none is already scheduled.
+    if conn.exists(_pending_key(using)) and conn.set(_lock_key(using), '1', nx=True, ex=LOCK_TTL):
+        _schedule(using)
+
+
+def drain_stranded(log=None):
+    """
+    Recover deferred work that no drain is scheduled to apply.
+
+    Scans for abandoned staging batches (a worker crashed or failed mid-apply)
+    and for pending hashes with no throttle lock (the drain meant to apply them
+    never ran). A no-op when there is nothing to recover, so the safety-net job
+    stays cheap.
+    """
+    log = log or logger
+    conn = _connection()
+
+    # Collect first, then apply: claiming a pending hash (RENAME) creates a new
+    # staging key, which must not be re-scanned as a stranded batch this pass.
+    targets = []
+    for raw_key in conn.scan_iter(match=f'{STAGING_PREFIX}*'):
+        staging = raw_key.decode() if isinstance(raw_key, bytes) else raw_key
+        targets.append((_alias_from_staging(staging), staging))
+
+    for raw_key in conn.scan_iter(match=f'{PENDING_PREFIX}*'):
+        pending = raw_key.decode() if isinstance(raw_key, bytes) else raw_key
+        using = pending[len(PENDING_PREFIX):]
+        if conn.exists(_lock_key(using)):
+            # A drain is already scheduled for this alias; leave it alone.
+            continue
+        staging = _claim(using)
+        if staging is not None:
+            targets.append((using, staging))
+
+    # Isolate failures: one batch that cannot be applied must not prevent the
+    # rest from being recovered. A failed batch is left staged for the next pass
+    # (its lease expires).
+    for using, staging in targets:
+        try:
+            _apply_staging(using, staging, log)
+        except Exception:
+            log.warning(f"Search cache: failed to apply deferred batch {staging}", exc_info=True)
+
+
+def _record_apply_failure(conn, batch_id):
+    """
+    Increment and return the failed-apply count for a staging batch, refreshing
+    its TTL so the counter cannot outlive the batch indefinitely.
+    """
+    key = _attempts_key(batch_id)
+    attempts = conn.incr(key)
+    conn.expire(key, ATTEMPTS_TTL)
+    return attempts
+
+
+def _apply_lock_key(db):
+    # A single advisory lock key per database/alias, shared by every background
+    # applier (the staged drain and the legacy SearchCacheJob path).
+    return f'netbox.search.apply:{db}'
+
+
+def _warn_if_stale(staging, log):
+    """
+    Log when a batch has been waiting far longer than a normal apply should take
+    (e.g. a database outage), so persistent staging growth is observable.
+    """
+    try:
+        claimed_at = int(staging.rsplit(':', 2)[1])
+    except (IndexError, ValueError):
+        return
+    age = int(time.time() - claimed_at)
+    if age > APPLY_LEASE:
+        log.warning(f"Search cache: applying a deferred batch claimed {age}s ago ({staging})")
+
+
+def _connection():
+    return get_connection(RQ_QUEUE_DEFAULT)
+
+
+def _field(object_type_id, pk):
+    return f'{object_type_id}:{pk}'
+
+
+def _pending_key(using):
+    return f'{PENDING_PREFIX}{using}'
+
+
+def _lock_key(using):
+    return f'{LOCK_PREFIX}{using}'
+
+
+def _lease_key(batch_id):
+    return f'{LEASE_PREFIX}{batch_id}'
+
+
+def _attempts_key(batch_id):
+    return f'{ATTEMPTS_PREFIX}{batch_id}'
+
+
+def _alias_from_staging(staging):
+    # 'netbox:search:draining:<using>:<epoch>:<batch_id>' -> '<using>'
+    return staging[len(STAGING_PREFIX):].rsplit(':', 2)[0]

netbox/netbox/search/jobs.py

@@ -1,6 +1,8 @@
 import logging
 
-from netbox.jobs import JobRunner
+from django.db import DEFAULT_DB_ALIAS
+
+from netbox.jobs import JobRunner, system_job
 from netbox.search.backends import search_backend
 
 # Internal search-indexing machinery; not part of the public/plugin API.
@@ -16,9 +18,45 @@ class SearchCacheJob(JobRunner):
         name = 'Search cache update'
 
     def run(self, using=None, cache_groups=None, remove_groups=None, **kwargs):
-        search_backend._apply_deferred_updates(
-            using=using,
-            cache_groups=cache_groups,
-            remove_groups=remove_groups,
-            log=self.logger,
-        )
+        using = using or DEFAULT_DB_ALIAS
+
+        # Back-compat: a job enqueued by a pre-coalescing version carried the
+        # dirty objects directly. Apply them as-is so in-flight legacy jobs keep
+        # working across a rolling upgrade. Serialize with a concurrent new-style
+        # drain via the same advisory lock so the two can't apply the same object
+        # concurrently (which would duplicate CachedValue rows). wait=True: a
+        # legacy job carries its own one-shot batch and cannot be retried, so it
+        # must not skip.
+        if cache_groups or remove_groups:
+            from django_pg_utils import advisory_lock
+
+            from netbox.search.deferred import _apply_lock_key
+            with advisory_lock(_apply_lock_key(using), wait=True, using=using):
+                search_backend._apply_deferred_updates(
+                    using=using,
+                    cache_groups=cache_groups,
+                    remove_groups=remove_groups,
+                    log=self.logger,
+                )
+            return
+
+        # Imported here to avoid a module-level circular import: deferred.py
+        # imports this module for SearchCacheJob.
+        from netbox.search import deferred
+        deferred.drain_and_apply(using, self.logger)
+
+
+@system_job(interval=5)
+class SearchCacheSafetyNetJob(JobRunner):
+    """
+    Periodic sweep that recovers deferred search-cache batches which were never
+    applied (e.g. the worker meant to drain them died, or application failed
+    part-way through). Retained staging data is retried; the batch is deleted
+    only once it has been applied successfully.
+    """
+    class Meta:
+        name = 'Search cache flush safety net'
+
+    def run(self, *args, **kwargs):
+        from netbox.search import deferred
+        deferred.drain_stranded(self.logger)

netbox/netbox/tests/test_search.py

@@ -1,5 +1,10 @@
+import os
+import subprocess
+import sys
+import time
 from unittest import mock
 
+from django.conf import settings
 from django.contrib.contenttypes.models import ContentType
 from django.db import DEFAULT_DB_ALIAS, ProgrammingError, connection, transaction
 from django.db.models.signals import post_delete, post_save
@@ -12,6 +17,7 @@ from extras.models import CachedValue
 from netbox.search import deferred
 from netbox.search.backends import SearchBackend, search_backend
 from netbox.search.jobs import SearchCacheJob
+from utilities.testing.mixins import RQQueueTestMixin
 
 
 def scheduled_search_flushes():
@@ -187,7 +193,7 @@ class SearchBackendTestCase(TestCase):
         self.assertEqual(len(results), 0)
 
 
-class DeferredCachingTestCase(TestCase):
+class DeferredCachingTestCase(RQQueueTestMixin, TestCase):
     """
     Tests for the deferred (post-commit) search caching machinery in
     netbox.search.deferred.
@@ -196,6 +202,11 @@ class DeferredCachingTestCase(TestCase):
     commit, so these tests assert on real CachedValue state and on the real
     per-transaction buffer (connection.run_on_commit) rather than mocking the
     queue.
+
+    Some tests patch the worker probe True and write to the real tasks Redis
+    (DB 0). RQQueueTestMixin flushes that database around each test and
+    serializes this class against the other Redis-backed suites, so a parallel
+    worker's flush can't wipe the coalescing keys mid-test.
     """
 
     def _scheduled_flush_aliases(self):
@@ -372,9 +383,11 @@ class DeferredCachingTestCase(TestCase):
 
     def test_flush_enqueues_job_when_worker_available(self):
         """
-        When an RQ worker is available, the flush enqueues a SearchCacheJob
-        carrying the dirty objects (and the originating database alias) rather
-        than indexing inline.
+        When an RQ worker is available, a committed batch is merged into the
+        shared Redis buffer and a single drain is scheduled, rather than indexing
+        inline. The originating database alias is carried on the scheduled job so
+        the deferred write targets the originating schema (e.g. a branch schema
+        under netbox-branching).
         """
         site_ct = ContentType.objects.get_for_model(Site)
 
@@ -386,15 +399,10 @@ class DeferredCachingTestCase(TestCase):
                     site = Site.objects.create(name='Enqueued', slug='enqueued')
 
         enqueue.assert_called_once()
-        kwargs = enqueue.call_args.kwargs
-        self.assertEqual(kwargs['cache_groups'], {site_ct.pk: [site.pk]})
-        self.assertEqual(kwargs['remove_groups'], {})
-        # The database alias captured from the post_save signal is forwarded to
-        # the job verbatim. This is what lets a deferred write target the
-        # originating schema (e.g. a branch schema under netbox-branching) even
-        # though the worker has no active routing context; cross-schema routing
-        # itself is covered by the netbox-branching test suite.
-        self.assertEqual(kwargs['using'], DEFAULT_DB_ALIAS)
+        self.assertEqual(enqueue.call_args.kwargs['using'], DEFAULT_DB_ALIAS)
+        # The object was buffered for a later drain rather than enqueued inline.
+        buffered = deferred._connection().hgetall(deferred._pending_key(DEFAULT_DB_ALIAS))
+        self.assertIn(f'{site_ct.pk}:{site.pk}', {k.decode() for k in buffered})
 
     def test_flush_falls_back_inline_when_broker_unreachable(self):
         """
@@ -452,6 +460,10 @@ class DeferredCachingTestCase(TestCase):
             CachedValue.objects.filter(object_type=content_type, object_id=site.pk).count(),
             len(SiteIndex.fields)
         )
+        # The scheduling lock is released so recovery can proceed promptly.
+        self.assertFalse(
+            deferred._connection().exists(deferred._lock_key(DEFAULT_DB_ALIAS))
+        )
 
     def test_cache_update_skips_deleted_object(self):
         """
@@ -569,6 +581,317 @@ class DeferredCachingTestCase(TestCase):
         self.assertIn(column, str(ctx.exception))
 
 
+class SearchCacheCoalescingTestCase(RQQueueTestMixin, TestCase):
+    """
+    Tests for the cross-transaction coalescing buffer in netbox.search.deferred.
+
+    These use the real tasks Redis connection (DB 0), so the RENAME/lock/lease
+    primitives are exercised for real rather than mocked. RQQueueTestMixin
+    flushes that database around each test and serializes this class against the
+    other Redis-backed suites; otherwise a parallel worker's flush would wipe the
+    coalescing keys mid-test.
+    """
+
+    @staticmethod
+    def _create_site(slug='coalesce'):
+        # Create the object, then discard any inline-indexed rows so tests start
+        # from a known-empty cache for it. Populate every indexed field so the
+        # expected CachedValue count is len(SiteIndex.fields).
+        site = Site.objects.create(
+            name=slug, slug=slug, facility='Alpha', description='desc',
+            physical_address='1 Test Way', shipping_address='1 Test Way',
+            comments='Lorem ipsum',
+        )
+        ct = ContentType.objects.get_for_model(Site)
+        CachedValue.objects.filter(object_type=ct, object_id=site.pk).delete()
+        return ct, site.pk
+
+    def _buffer(self, batch):
+        with mock.patch('netbox.search.deferred.any_workers_for_queue', return_value=True):
+            with mock.patch.object(SearchCacheJob, 'enqueue'):
+                deferred._flush(batch, DEFAULT_DB_ALIAS)
+
+    def test_many_committed_transactions_coalesce_into_one_job(self):
+        """
+        Regression guard for #23259: many separate committed transactions must
+        coalesce into a single scheduled job, not one job per transaction.
+        """
+        ct, pk = self._create_site()
+
+        with mock.patch('netbox.search.deferred.any_workers_for_queue', return_value=True):
+            with mock.patch.object(SearchCacheJob, 'enqueue') as enqueue:
+                for i in range(25):
+                    deferred._flush({(ct.pk, pk + i): deferred.OP_CACHE}, DEFAULT_DB_ALIAS)
+
+        enqueue.assert_called_once()
+        buffered = deferred._connection().hgetall(deferred._pending_key(DEFAULT_DB_ALIAS))
+        self.assertEqual(len(buffered), 25)
+
+    def test_later_op_wins_in_buffer(self):
+        ct, pk = self._create_site()
+        self._buffer({(ct.pk, pk): deferred.OP_CACHE})
+        self._buffer({(ct.pk, pk): deferred.OP_REMOVE})
+        buffered = deferred._connection().hgetall(deferred._pending_key(DEFAULT_DB_ALIAS))
+        self.assertEqual(buffered[f'{ct.pk}:{pk}'.encode()], b'remove')
+
+    def test_drain_applies_and_clears_buffer(self):
+        ct, pk = self._create_site()
+        self._buffer({(ct.pk, pk): deferred.OP_CACHE})
+        deferred.drain_and_apply(DEFAULT_DB_ALIAS)
+        self.assertEqual(
+            CachedValue.objects.filter(object_type=ct, object_id=pk).count(),
+            len(SiteIndex.fields),
+        )
+        conn = deferred._connection()
+        self.assertFalse(conn.exists(deferred._pending_key(DEFAULT_DB_ALIAS)))
+        self.assertEqual(list(conn.scan_iter(match=f'{deferred.STAGING_PREFIX}*')), [])
+
+    def test_drain_removes_and_clears(self):
+        ct, pk = self._create_site()
+        search_backend.cache(Site.objects.get(pk=pk))
+        self.assertTrue(CachedValue.objects.filter(object_type=ct, object_id=pk).exists())
+        self._buffer({(ct.pk, pk): deferred.OP_REMOVE})
+        deferred.drain_and_apply(DEFAULT_DB_ALIAS)
+        self.assertFalse(CachedValue.objects.filter(object_type=ct, object_id=pk).exists())
+
+    def test_second_claim_returns_none(self):
+        ct, pk = self._create_site()
+        conn = deferred._connection()
+        conn.hset(deferred._pending_key(DEFAULT_DB_ALIAS), mapping={f'{ct.pk}:{pk}': 'cache'})
+        self.assertIsNotNone(deferred._claim(DEFAULT_DB_ALIAS))
+        self.assertIsNone(deferred._claim(DEFAULT_DB_ALIAS))
+
+    def test_update_arriving_during_drain_is_rescheduled(self):
+        ct, pk = self._create_site()
+        conn = deferred._connection()
+        conn.hset(deferred._pending_key(DEFAULT_DB_ALIAS), mapping={f'{ct.pk}:{pk}': 'cache'})
+
+        def apply_and_arrive(*args, **kwargs):
+            # Simulate a commit landing while the claimed batch is being applied.
+            conn.hset(deferred._pending_key(DEFAULT_DB_ALIAS), mapping={f'{ct.pk}:{pk}': 'remove'})
+
+        with mock.patch.object(search_backend, '_apply_deferred_updates', side_effect=apply_and_arrive):
+            with mock.patch.object(SearchCacheJob, 'enqueue') as enqueue:
+                deferred.drain_and_apply(DEFAULT_DB_ALIAS)
+
+        enqueue.assert_called_once()
+        self.assertTrue(conn.exists(deferred._pending_key(DEFAULT_DB_ALIAS)))
+
+    def test_stranded_staging_is_recovered(self):
+        ct, pk = self._create_site()
+        conn = deferred._connection()
+        staging = f'{deferred.STAGING_PREFIX}{DEFAULT_DB_ALIAS}:{int(time.time())}:deadbeef'
+        conn.hset(staging, mapping={f'{ct.pk}:{pk}': 'cache'})
+
+        deferred.drain_stranded()
+
+        self.assertEqual(
+            CachedValue.objects.filter(object_type=ct, object_id=pk).count(),
+            len(SiteIndex.fields),
+        )
+        self.assertFalse(conn.exists(staging))
+
+    def test_failed_apply_retains_staging_and_retry_converges(self):
+        ct, pk = self._create_site()
+        conn = deferred._connection()
+        staging = f'{deferred.STAGING_PREFIX}{DEFAULT_DB_ALIAS}:{int(time.time())}:cafe'
+        conn.hset(staging, mapping={f'{ct.pk}:{pk}': 'cache'})
+
+        with mock.patch.object(
+            search_backend, '_apply_deferred_updates', side_effect=ProgrammingError('boom')
+        ):
+            deferred.drain_stranded()
+
+        # The only copy is retained on failure, never discarded.
+        self.assertTrue(conn.exists(staging))
+
+        # Simulate lease expiry, then retry for real: the retry converges.
+        for raw_key in conn.scan_iter(match=f'{deferred.LEASE_PREFIX}*'):
+            conn.delete(raw_key)
+        deferred.drain_stranded()
+
+        self.assertEqual(
+            CachedValue.objects.filter(object_type=ct, object_id=pk).count(),
+            len(SiteIndex.fields),
+        )
+        self.assertFalse(conn.exists(staging))
+
+    def test_advisory_lock_contention_skips_apply(self):
+        """
+        If another applier holds the advisory lock (its Redis lease may have
+        expired while it is still running), a second applier must not apply the
+        same batch concurrently -- that is the #23259 duplicate-row path.
+        """
+        ct, pk = self._create_site()
+        conn = deferred._connection()
+        staging = f'{deferred.STAGING_PREFIX}{DEFAULT_DB_ALIAS}:{int(time.time())}:1ease'
+        conn.hset(staging, mapping={f'{ct.pk}:{pk}': 'cache'})
+        batch_id = staging.rsplit(':', 1)[1]
+
+        with mock.patch('netbox.search.deferred.advisory_lock') as advisory:
+            advisory.return_value.__enter__.return_value = False
+            with mock.patch.object(search_backend, '_apply_deferred_updates') as apply:
+                deferred._apply_staging(DEFAULT_DB_ALIAS, staging, deferred.logger)
+                apply.assert_not_called()
+
+        # The batch is retained for the current lock holder (or the safety net),
+        # and contention is not counted as an apply failure.
+        self.assertTrue(conn.exists(staging))
+        self.assertFalse(conn.exists(deferred._attempts_key(batch_id)))
+
+    def test_repeated_apply_failure_is_capped_and_batch_dropped(self):
+        """
+        A batch that keeps failing is retained below the cap, then dropped at the
+        cap with an error-level log instead of blocking indexing of unrelated
+        objects forever.
+        """
+        ct, pk = self._create_site()
+        conn = deferred._connection()
+        staging = f'{deferred.STAGING_PREFIX}{DEFAULT_DB_ALIAS}:{int(time.time())}:cafe2'
+        conn.hset(staging, mapping={f'{ct.pk}:{pk}': 'cache'})
+        batch_id = staging.rsplit(':', 1)[1]
+
+        with mock.patch.object(
+            search_backend, '_apply_deferred_updates', side_effect=ProgrammingError('boom')
+        ):
+            with self.assertLogs('netbox.search.deferred', level='ERROR') as logs:
+                for _ in range(deferred.MAX_APPLY_ATTEMPTS - 1):
+                    # Simulate the lease expiring between safety-net passes.
+                    conn.delete(deferred._lease_key(batch_id))
+                    with self.assertRaises(ProgrammingError):
+                        deferred._apply_staging(DEFAULT_DB_ALIAS, staging, deferred.logger)
+                    self.assertTrue(conn.exists(staging))
+
+                conn.delete(deferred._lease_key(batch_id))
+                deferred._apply_staging(DEFAULT_DB_ALIAS, staging, deferred.logger)
+
+        self.assertFalse(conn.exists(staging))
+        self.assertFalse(conn.exists(deferred._attempts_key(batch_id)))
+        self.assertTrue(any('dropping deferred batch' in m for m in logs.output))
+
+    def test_successful_apply_clears_failure_counter(self):
+        ct, pk = self._create_site()
+        conn = deferred._connection()
+        staging = f'{deferred.STAGING_PREFIX}{DEFAULT_DB_ALIAS}:{int(time.time())}:f00d'
+        conn.hset(staging, mapping={f'{ct.pk}:{pk}': 'cache'})
+        batch_id = staging.rsplit(':', 1)[1]
+        conn.set(deferred._attempts_key(batch_id), 3)
+
+        deferred._apply_staging(DEFAULT_DB_ALIAS, staging, deferred.logger)
+
+        self.assertFalse(conn.exists(staging))
+        self.assertFalse(conn.exists(deferred._attempts_key(batch_id)))
+
+    def test_lease_prevents_concurrent_apply(self):
+        ct, pk = self._create_site()
+        conn = deferred._connection()
+        staging = f'{deferred.STAGING_PREFIX}{DEFAULT_DB_ALIAS}:{int(time.time())}:feed'
+        conn.hset(staging, mapping={f'{ct.pk}:{pk}': 'cache'})
+        conn.set(deferred._lease_key('feed'), '1', nx=True, ex=60)
+
+        with mock.patch.object(search_backend, '_apply_deferred_updates') as apply:
+            deferred._apply_staging(DEFAULT_DB_ALIAS, staging, deferred.logger)
+            apply.assert_not_called()
+
+        self.assertTrue(conn.exists(staging))
+
+    def test_expired_lease_allows_takeover(self):
+        ct, pk = self._create_site()
+        conn = deferred._connection()
+        staging = f'{deferred.STAGING_PREFIX}{DEFAULT_DB_ALIAS}:{int(time.time())}:beef'
+        conn.hset(staging, mapping={f'{ct.pk}:{pk}': 'cache'})
+        # A lease is set then expires before the batch is applied.
+        conn.set(deferred._lease_key('beef'), '1', nx=True, ex=60)
+        conn.delete(deferred._lease_key('beef'))
+
+        deferred._apply_staging(DEFAULT_DB_ALIAS, staging, deferred.logger)
+
+        self.assertEqual(
+            CachedValue.objects.filter(object_type=ct, object_id=pk).count(),
+            len(SiteIndex.fields),
+        )
+        self.assertFalse(conn.exists(staging))
+
+    def test_draining_one_batch_does_not_delete_newer_pending(self):
+        ct, pk = self._create_site()
+        conn = deferred._connection()
+        staging = f'{deferred.STAGING_PREFIX}{DEFAULT_DB_ALIAS}:{int(time.time())}:abcd'
+        conn.hset(staging, mapping={f'{ct.pk}:{pk}': 'cache'})
+        # A newer commit arrives after the batch was claimed.
+        conn.hset(deferred._pending_key(DEFAULT_DB_ALIAS), mapping={f'{ct.pk}:{pk + 1}': 'remove'})
+
+        with mock.patch.object(SearchCacheJob, 'enqueue'):
+            deferred._apply_staging(DEFAULT_DB_ALIAS, staging, deferred.logger)
+
+        buffered = conn.hgetall(deferred._pending_key(DEFAULT_DB_ALIAS))
+        self.assertIn(f'{ct.pk}:{pk + 1}'.encode(), buffered)
+        self.assertFalse(conn.exists(staging))
+
+    def test_safety_net_skips_alias_with_scheduled_drain(self):
+        ct, pk = self._create_site()
+        conn = deferred._connection()
+        conn.hset(deferred._pending_key(DEFAULT_DB_ALIAS), mapping={f'{ct.pk}:{pk}': 'cache'})
+        conn.set(deferred._lock_key(DEFAULT_DB_ALIAS), '1', nx=True, ex=60)
+
+        with mock.patch.object(search_backend, '_apply_deferred_updates') as apply:
+            deferred.drain_stranded()
+            apply.assert_not_called()
+
+        self.assertTrue(conn.exists(deferred._pending_key(DEFAULT_DB_ALIAS)))
+
+    def test_safety_net_noop_when_empty(self):
+        with mock.patch.object(search_backend, '_apply_deferred_updates') as apply:
+            deferred.drain_stranded()
+            apply.assert_not_called()
+
+    def test_legacy_job_arguments_still_apply(self):
+        """
+        A SearchCacheJob enqueued by a pre-coalescing version carries cache_groups
+        / remove_groups directly; it must still be applied (rolling upgrade).
+        """
+        ct, pk = self._create_site()
+        SearchCacheJob.run(
+            mock.Mock(),
+            using=DEFAULT_DB_ALIAS,
+            cache_groups={ct.pk: [pk]},
+            remove_groups={},
+        )
+        self.assertEqual(
+            CachedValue.objects.filter(object_type=ct, object_id=pk).count(),
+            len(SiteIndex.fields),
+        )
+
+    def test_safety_net_job_registered_by_app_ready(self):
+        """
+        The safety-net job must be registered by CoreConfig.ready() importing
+        netbox.search.jobs -- not merely by this test module's own import of that
+        module, which would make an in-process assertion vacuous. Boot Django in
+        a clean interpreter that never imports netbox.search.jobs explicitly and
+        inspect the registry there.
+        """
+        script = (
+            "import django; django.setup();"
+            "from netbox.registry import registry;"
+            "jobs = [c for c in registry['system_jobs']"
+            " if getattr(c, 'name', None) == 'Search cache flush safety net'];"
+            "assert len(jobs) == 1, jobs;"
+            "assert registry['system_jobs'][jobs[0]]['interval'] == 5, jobs;"
+            "print('REGISTERED')"
+        )
+        env = os.environ.copy()
+        env.setdefault('DJANGO_SETTINGS_MODULE', 'netbox.settings')
+        env.setdefault('NETBOX_CONFIGURATION', 'netbox.configuration_testing')
+        result = subprocess.run(
+            [sys.executable, '-c', script],
+            cwd=str(settings.BASE_DIR),
+            env=env,
+            capture_output=True,
+            text=True,
+        )
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn('REGISTERED', result.stdout)
+
+
 class AutocommitCachingTestCase(TransactionTestCase):
     """
     Tests for the synchronous (autocommit) indexing path. Uses TransactionTestCase