From 6f3efea1d9fb4a8e413eb4cdbeb40a15b2406f73 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 8 Jul 2026 12:21:52 +0000 Subject: [PATCH 1/2] Send _get_server_keys_json invalidations over replication as JSON MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The cache key of `_get_server_keys_json` is a single argument which is itself a `(server_name, key_id)` tuple, and `store_server_keys_response` passed that nested tuple straight into the cache invalidation stream. psycopg2 quietly serialises the inner tuple as a Postgres *record*, so the `keys` column of `cache_invalidation_stream_by_instance` ended up holding the record literal as a single string (e.g. `{"(srv,ed25519:abc)"}`) — which never matches the real cache key on the receiving side, i.e. the invalidation has always been a silent no-op on workers. The native Rust backend's stricter parameter conversion turns the same nested tuple into a loud `TypeError: unsupported parameter type for postgres: tuple`. Fix it the same way #18899 did for `_get_e2e_cross_signing_signatures_for_device`, which has the same nested-tuple key shape: invalidate the local cache directly, JSON-encode the key for the replication row, and decode it again in `process_replication_rows`. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_019ZdvuLkg7Lm7wDtPQzDnJx --- synapse/storage/databases/main/cache.py | 22 ++++++++++ synapse/storage/databases/main/keys.py | 16 +++++-- tests/storage/databases/main/test_cache.py | 51 ++++++++++++++++++++++ 3 files changed, 86 insertions(+), 3 deletions(-) diff --git a/synapse/storage/databases/main/cache.py b/synapse/storage/databases/main/cache.py index a4530796f20..45fe605d8bc 100644 --- a/synapse/storage/databases/main/cache.py +++ b/synapse/storage/databases/main/cache.py @@ -70,6 +70,10 @@ "_get_e2e_cross_signing_signatures_for_device" ) +# As above: this cache takes a single argument which is itself a tuple, which +# requires special handling. +GET_SERVER_KEYS_JSON_CACHE_NAME = "_get_server_keys_json" + # How long between cache invalidation table cleanups, once we have caught up # with the backlog. REGULAR_CLEANUP_INTERVAL = Duration(hours=1) @@ -305,6 +309,24 @@ def process_replication_rows( self._get_e2e_cross_signing_signatures_for_device.invalidate( # type: ignore[attr-defined] ((user_id, device_id),) ) + elif row.cache_func == GET_SERVER_KEYS_JSON_CACHE_NAME: + # As above: each entry in "keys" is a JSON-encoded + # (server_name, key_id) pair, since the cache takes a single + # argument which is itself a tuple and we cannot send nested + # information over replication. + for json_str in row.keys: + try: + server_name, key_id = json.loads(json_str) + except (json.JSONDecodeError, TypeError, ValueError): + logger.error( + "Failed to deserialise cache key as valid JSON: %s", + json_str, + ) + continue + + self._get_server_keys_json.invalidate( # type: ignore[attr-defined] + ((server_name, key_id),) + ) else: self._attempt_to_invalidate_cache(row.cache_func, row.keys) diff --git a/synapse/storage/databases/main/keys.py b/synapse/storage/databases/main/keys.py index f81257b5a1f..3e733643e76 100644 --- a/synapse/storage/databases/main/keys.py +++ b/synapse/storage/databases/main/keys.py @@ -113,10 +113,20 @@ def store_server_keys_response_txn(txn: LoggingTransaction) -> None: # invalidate takes a tuple corresponding to the params of # _get_server_keys_json. _get_server_keys_json only takes one # param, which is itself the 2-tuple (server_name, key_id). - self._invalidate_cache_and_stream_bulk( + # + # Invalidate the local cache directly, but we can only send + # primitive types per argument over replication, so JSON-encode the + # nested key and unpack it on the receiving side (see + # `CacheInvalidationWorkerStore.process_replication_rows`). + for key_id in verify_keys: + txn.call_after( + self._get_server_keys_json.invalidate, + ((server_name, key_id),), + ) + self._send_invalidation_to_replication_bulk( txn, - self._get_server_keys_json, - [((server_name, key_id),) for key_id in verify_keys], + self._get_server_keys_json.__name__, + [(json.dumps([server_name, key_id]),) for key_id in verify_keys], ) self._invalidate_cache_and_stream_bulk( txn, diff --git a/tests/storage/databases/main/test_cache.py b/tests/storage/databases/main/test_cache.py index 3900a352901..1e7ae7e5a31 100644 --- a/tests/storage/databases/main/test_cache.py +++ b/tests/storage/databases/main/test_cache.py @@ -20,7 +20,10 @@ # from unittest.mock import Mock, call +from signedjson.key import generate_signing_key, get_verify_key + from synapse.storage.database import LoggingTransaction +from synapse.storage.keys import FetchKeyResult from tests.replication._base import BaseMultiWorkerStreamTestCase from tests.unittest import HomeserverTestCase @@ -122,3 +125,51 @@ def test_txn(txn: LoggingTransaction) -> None: [call(key_list) for key_list in keys_to_invalidate], any_order=True, ) + + def test_server_keys_json_invalidation_replicates(self) -> None: + """`_get_server_keys_json` takes a single argument which is itself a + tuple, and we can't send nested keys over replication: the keys are + JSON-encoded on the sending side and decoded again in + `process_replication_rows`. Check the round trip invalidates the right + cache entry on the worker. + """ + master_invalidate = Mock() + worker_invalidate = Mock() + + self.store._get_server_keys_json.invalidate = master_invalidate + worker = self.make_worker_hs("synapse.app.generic_worker") + worker_ds = worker.get_datastores().main + worker_ds._get_server_keys_json.invalidate = worker_invalidate + + signing_key = generate_signing_key("ver1") + verify_key = get_verify_key(signing_key) + key_id = f"{verify_key.alg}:{verify_key.version}" + + assert self.store._cache_id_gen is not None + initial_token = self.store._cache_id_gen.get_current_token() + + self.get_success( + self.store.store_server_keys_response( + "server1", + from_server="server2", + ts_added_ms=self.clock.time_msec(), + verify_keys={ + key_id: FetchKeyResult( + verify_key=verify_key, valid_until_ts=200_000 + ), + }, + response_json={}, + ) + ) + second_token = self.store._cache_id_gen.get_current_token() + self.assertGreater(second_token, initial_token) + + self.get_success( + worker.get_replication_data_handler().wait_for_stream_position( + "master", "caches", second_token + ) + ) + + expected_key = (("server1", key_id),) + master_invalidate.assert_called_once_with(expected_key) + worker_invalidate.assert_called_once_with(expected_key) From 6f0225969be809455d2f57a3ab750910ccaf69b2 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 15 Jul 2026 15:43:20 +0100 Subject: [PATCH 2/2] Newsfile --- changelog.d/19966.bugfix | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/19966.bugfix diff --git a/changelog.d/19966.bugfix b/changelog.d/19966.bugfix new file mode 100644 index 00000000000..5940159f568 --- /dev/null +++ b/changelog.d/19966.bugfix @@ -0,0 +1 @@ +Fix server key cache invalidations being silently dropped on workers.