Fix _get_server_keys_json invalidations over replication as JSON (#19966)

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 https://github.com/element-hq/synapse/pull/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`.

Found as part of the the effort to port the database pool to Rust.

changelog.d/19966.bugfix

@@ -0,0 +1 @@
+Fix server key cache invalidations being silently dropped on workers.

synapse/storage/databases/main/cache.py

@@ -70,6 +70,10 @@ GET_E2E_CROSS_SIGNING_SIGNATURES_FOR_DEVICE_CACHE_NAME = (
     "_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 @@ class CacheInvalidationWorkerStore(SQLBaseStore):
                         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)
 

synapse/storage/databases/main/keys.py

@@ -113,10 +113,20 @@ class KeyStore(CacheInvalidationWorkerStore):
             # 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,

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 @@ class CacheInvalidationOverReplicationTestCase(BaseMultiWorkerStreamTestCase):
             [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)