diff --git a/netbox/netbox/search/backends.py b/netbox/netbox/search/backends.py index 08fb12bdf..43bc9bb31 100644 --- a/netbox/netbox/search/backends.py +++ b/netbox/netbox/search/backends.py @@ -349,23 +349,63 @@ class CachedValueSearchBackend(SearchBackend): return self._remove_by_id(object_type.pk, [instance.pk], using=using) - # Postgres SQLSTATEs indicating the target schema/table no longer exists. This happens when a - # branch is merged or deprovisioned (its schema dropped) between the time an update was enqueued - # and when it is applied. Such errors are expected and safe to skip; the index is rebuilt on the - # next reindex. Any other DatabaseError (e.g. a deadlock or lost connection) is transient and must - # propagate so the work fails visibly, rather than silently dropping index updates. - _MISSING_SCHEMA_SQLSTATES = frozenset(( + # Postgres SQLSTATEs indicating this backend's own CachedValue table (or the schema it lives in) + # no longer exists. This happens when a branch is merged or deprovisioned (its schema, and every + # table in it including its copy of CachedValue, dropped) between the time an update was enqueued + # and when it is applied. There is nothing to write to and nothing worth retrying -- a later write + # for a surviving branch's CachedValue table is unaffected -- so this is expected and safe to + # skip. Any other DatabaseError (e.g. a deadlock, lost connection, or CachedValue itself being out + # of sync with a not-yet-migrated deployment) is not expected and must propagate so the work fails + # visibly, rather than silently dropping index updates. This set applies to every write this + # backend makes to CachedValue (removals below, and the remove+insert inside the cache loop) -- + # deliberately not to reads of the model being indexed; see _STALE_INDEX_TARGET_SQLSTATES for + # that. + _MISSING_CACHE_TABLE_SQLSTATES = frozenset(( '3F000', # invalid_schema_name '42P01', # undefined_table )) - def _is_missing_schema(self, exc): + # Additionally tolerated when reading the model being (re)indexed -- not when writing to + # CachedValue (see _MISSING_CACHE_TABLE_SQLSTATES above, which this extends). A plugin whose + # models are dynamically regenerated per branch (e.g. netbox-custom-objects) resolves + # ObjectType.model_class() to a branch-unaware class pinned to main; if a branch has renamed or + # removed a column since an update was enqueued, that stale class's column no longer matches the + # branch's live table, and reading it raises undefined_column rather than undefined_table. + # Unlike a dropped schema, this is *not* self-healing on "the next reindex": model_class() + # resolves the same stale, main-pinned class every time, so the object stays unindexed until the + # branch is merged or reverted. Scoping this to the read only -- rather than folding it into + # _MISSING_CACHE_TABLE_SQLSTATES broadly -- keeps a genuine defect in the write path (e.g. code + # deployed against a database that has not yet been migrated, which would also raise + # undefined_column) from being silently downgraded to a warning. + # + # This covers only the top-level read of the model's own columns, not a related object a search + # index's to_cache() might lazily traverse into (e.g. a relational field indexed with a positive + # search_weight): such a traversal issues its own query inside the write step below, which does + # not tolerate undefined_column. A plugin whose indexed fields never reference another of its own + # dynamically-regenerated models is unaffected; one that does would need a plugin-side fix (making + # ObjectType.model_class() branch-aware) rather than a wider exemption here. + _STALE_INDEX_TARGET_SQLSTATES = _MISSING_CACHE_TABLE_SQLSTATES | frozenset(( + '42703', # undefined_column + )) + + def _is_missing_cache_table(self, exc): """ - Return True if the given DatabaseError was caused by the target schema/table no longer existing - (vs. a transient error that should propagate). + Return True if the given DatabaseError was caused by CachedValue's own schema/table no longer + existing (vs. a transient error that should propagate). Covers writes to CachedValue; see + _is_stale_index_target() for the read side of a deferred update. """ sqlstate = getattr(getattr(exc, '__cause__', None), 'sqlstate', None) - return sqlstate in self._MISSING_SCHEMA_SQLSTATES + return sqlstate in self._MISSING_CACHE_TABLE_SQLSTATES + + def _is_stale_index_target(self, exc): + """ + Return True if the given DatabaseError was caused by the model being (re)indexed no longer + matching what was expected when the update was enqueued -- its schema or table (as for + _is_missing_cache_table()), or, for a model whose columns can themselves diverge per branch, + an individual column. + """ + sqlstate = getattr(getattr(exc, '__cause__', None), 'sqlstate', None) + return sqlstate in self._STALE_INDEX_TARGET_SQLSTATES def _apply_deferred_updates(self, using=None, cache_groups=None, remove_groups=None, log=logger): """ @@ -377,17 +417,27 @@ class CachedValueSearchBackend(SearchBackend): written to the originating database/schema (e.g. a branch schema under netbox-branching), regardless of any routing context that is no longer active by the time this runs. """ - # Removals are a single DELETE per content type, so (unlike the cache loop below) there is no - # multi-step state to wrap in a transaction. A transient error here propagates and errors the - # caller; the remaining work is dropped rather than retried (NetBox does not retry these jobs - # by default) and is recovered by the next reindex. for object_type_id, pks in (remove_groups or {}).items(): + # Resolved once and reused for both the atomic() block and the delete below: passing + # `using` straight through when it's falsy would let transaction.atomic() default to + # DEFAULT_DB_ALIAS while _remove_by_id() defers to the router (see cache()'s own comment + # on this) -- the savepoint would then belong to a different connection than the one the + # DELETE actually runs on, on any deployment where CachedValue is routed elsewhere. + db = using or CachedValue.objects.db try: - self._remove_by_id(object_type_id, pks, using=using) + # A tolerated DatabaseError still needs a savepoint to roll back to: without one, + # Postgres leaves the connection's enclosing transaction aborted (refusing every + # further statement in it until a rollback) even though the Python exception was + # caught, breaking every later iteration of this loop -- not just this one. + with transaction.atomic(using=db): + self._remove_by_id(object_type_id, pks, using=db) except DatabaseError as e: - if not self._is_missing_schema(e): + if not self._is_missing_cache_table(e): raise - log.warning(f"Skipping search cache removal for object type {object_type_id}: {e}") + log.warning( + f"Skipping search cache removal for object type {object_type_id}: " + f"CachedValue's own table or schema no longer exists ({e})" + ) for object_type_id, pks in (cache_groups or {}).items(): try: @@ -398,22 +448,54 @@ class CachedValueSearchBackend(SearchBackend): if model is None: continue - try: - # Re-fetch live instances from the originating database. Reading on `using` is - # required: a branch object's PK may be absent (or refer to a different object) on the - # default connection. - queryset = model.objects.using(using).filter(pk__in=pks) + # Resolved once per group, for the same reason as the removal loop above -- and kept + # separate for the read vs. the write, since a router could place the indexed model and + # CachedValue on different aliases (in which case no single atomic() spans both anyway; + # see the outer/inner split below). + read_db = using or model._default_manager.db + write_db = using or CachedValue.objects.db - # Clear any stale entries for these objects, then re-insert. Wrapping both in one - # transaction avoids leaving an object with no cache rows if execution fails between - # the delete and the insert. - with transaction.atomic(using=using): - self._remove_by_id(object_type_id, pks, using=using) - self.cache(queryset, remove_existing=False, using=using) + try: + # The outer atomic() makes the delete+insert pair below a single write. It does + # *not* give the read a consistent snapshot with that write -- PostgreSQL's default + # (and NetBox's) READ COMMITTED isolation takes a fresh snapshot per statement + # regardless of transaction boundaries, so a concurrent update can still land + # between the read and the write either way. That race is pre-existing and benign: + # the concurrent save schedules its own deferred update, so the object is reindexed + # again regardless of which value this pass happened to write. + # + # Nested inside it, the read gets its own savepoint so a tolerated failure there + # (see _is_stale_index_target(), which tolerates a wider set of SQLSTATES than a + # write to CachedValue does) rolls back only the read, without ever reaching -- or + # needing to roll back -- the write. + with transaction.atomic(using=write_db): + try: + with transaction.atomic(using=read_db): + # Reading on `read_db` is required: a branch object's PK may be absent + # (or refer to a different object) on the default connection. + instances = list(model.objects.using(read_db).filter(pk__in=pks)) + except DatabaseError as e: + if not self._is_stale_index_target(e): + raise + log.warning( + f"Skipping search cache update for object type {object_type_id}: the " + f"indexed model no longer matches its table, e.g. a branch-diverged " + f"column ({e})" + ) + continue + + # Clear any stale entries for these objects, then re-insert. Wrapping both in + # one transaction avoids leaving an object with no cache rows if execution fails + # between the delete and the insert. + self._remove_by_id(object_type_id, pks, using=write_db) + self.cache(instances, remove_existing=False, using=write_db) except DatabaseError as e: - if not self._is_missing_schema(e): + if not self._is_missing_cache_table(e): raise - log.warning(f"Skipping search cache update for object type {object_type_id}: {e}") + log.warning( + f"Skipping search cache update for object type {object_type_id}: " + f"CachedValue's own table or schema no longer exists ({e})" + ) def clear(self, object_types=None): qs = CachedValue.objects.all() diff --git a/netbox/netbox/tests/test_search.py b/netbox/netbox/tests/test_search.py index 47467c8fa..4c9d0c5d4 100644 --- a/netbox/netbox/tests/test_search.py +++ b/netbox/netbox/tests/test_search.py @@ -1,12 +1,12 @@ from unittest import mock from django.contrib.contenttypes.models import ContentType -from django.db import DEFAULT_DB_ALIAS, connection, transaction +from django.db import DEFAULT_DB_ALIAS, ProgrammingError, connection, transaction from django.db.models.signals import post_delete, post_save from django.test import TestCase, TransactionTestCase from redis.exceptions import ConnectionError as RedisConnectionError -from dcim.models import Site +from dcim.models import Manufacturer, Site from dcim.search import SiteIndex from extras.models import CachedValue from netbox.search import deferred @@ -471,6 +471,103 @@ class DeferredCachingTestCase(TestCase): CachedValue.objects.filter(object_type=site_ct, object_id=pk).exists() ) + def test_remove_groups_skips_missing_cache_table(self): + """ + Regression test for the per-iteration transaction.atomic() in the remove_groups loop: + without a savepoint to roll back to, a tolerated failure on the first group's DELETE + leaves the connection's transaction aborted, and a second group's DELETE in the same call + then dies with 25P02 (in_failed_sql_transaction) -- a SQLSTATE in neither tolerated set -- + instead of being tolerated like the first. Two remove_groups entries are required to + observe this: a single entry never reaches a second DELETE to fail. + """ + table = CachedValue._meta.db_table + with connection.cursor() as cursor: + cursor.execute('SET CONSTRAINTS ALL IMMEDIATE') + cursor.execute(f'ALTER TABLE {table} RENAME TO {table}_missing') + + # Neither group's object type/pk needs to correspond to a real object: _remove_by_id() + # only uses them to filter CachedValue, which is what no longer exists here. + search_backend._apply_deferred_updates( + using=None, cache_groups={}, remove_groups={1: [1], 2: [2]}, + ) + + def test_cache_update_skips_dropped_column(self): + """ + _apply_deferred_updates tolerates the model being indexed having lost a column the ORM + still expects (e.g. a plugin's dynamically-generated model whose field was renamed or + removed within an unmerged branch since the update was enqueued): it must not error, and + a cache entry already written by an earlier, successful index must survive untouched. + The read of the model being indexed happens before any write to CachedValue, so a + tolerated failure there means the removal and re-insert are never even attempted for + this object -- that ordering, not a rollback, is what leaves the earlier entry + untouched, and is the property actually worth protecting here, not just "no exception" + (netboxlabs/netbox-custom-objects#651; netbox-community/netbox#22909). + + Manufacturer (rather than Site, used elsewhere in this file) is used here because it + carries no denormalization triggers of its own -- but every FK constraint Django creates + on PostgreSQL is DEFERRABLE INITIALLY DEFERRED by default, so this test's own INSERT + (via its object_type FK) leaves a pending RI trigger event within the same transaction; + Postgres refuses to ALTER TABLE until that settles, hence the explicit SET CONSTRAINTS + below. + """ + manufacturer = Manufacturer.objects.create( + name='Column Dropped', slug='column-dropped', description='doomed', + ) + manufacturer_ct = ContentType.objects.get_for_model(Manufacturer) + pk = manufacturer.pk + + # Seed a real, successful index entry -- the thing that must survive intact below. + search_backend.cache(manufacturer) + cached_before = set( + CachedValue.objects.filter(object_type=manufacturer_ct, object_id=pk) + .values_list('field', 'value') + ) + self.assertTrue(cached_before, 'setup failed to index the object at all') + + column = Manufacturer._meta.get_field('description').column + with connection.cursor() as cursor: + cursor.execute('SET CONSTRAINTS ALL IMMEDIATE') + cursor.execute(f'ALTER TABLE {Manufacturer._meta.db_table} DROP COLUMN {column}') + + # No exception, and the prior index entry is left exactly as it was: a tolerated failure + # must not wipe out previously-good data. + search_backend._apply_deferred_updates( + using=None, cache_groups={manufacturer_ct.pk: [pk]}, remove_groups={}, + ) + cached_after = set( + CachedValue.objects.filter(object_type=manufacturer_ct, object_id=pk) + .values_list('field', 'value') + ) + self.assertEqual(cached_before, cached_after) + + def test_cache_update_propagates_unrelated_database_error(self): + """ + The tolerance added for a stale target column (test_cache_update_skips_dropped_column) is + scoped to reading the model being indexed. A DatabaseError raised while writing to this + backend's own CachedValue table -- e.g. an undefined_column there too, which would + indicate a genuine defect (such as code deployed against a database that has not yet + been migrated) rather than a plugin's branch-diverged schema -- must still propagate + rather than being swallowed by the widened tolerance. + """ + manufacturer = Manufacturer.objects.create(name='Propagates', slug='propagates') + manufacturer_ct = ContentType.objects.get_for_model(Manufacturer) + + column = CachedValue._meta.get_field('weight').column + with connection.cursor() as cursor: + cursor.execute('SET CONSTRAINTS ALL IMMEDIATE') + cursor.execute(f'ALTER TABLE {CachedValue._meta.db_table} DROP COLUMN {column}') + + # Pin down that this specifically is the write to CachedValue failing, not some + # unrelated ProgrammingError -- otherwise a future change that broke the read side + # instead could still satisfy this assertion. + with self.assertRaises(ProgrammingError) as ctx: + search_backend._apply_deferred_updates( + using=None, + cache_groups={manufacturer_ct.pk: [manufacturer.pk]}, + remove_groups={}, + ) + self.assertIn(column, str(ctx.exception)) + class AutocommitCachingTestCase(TransactionTestCase): """