Update `get_pdu` to return the original, pristine `EventBase` (#13320)

Update `get_pdu` to return the untouched, pristine `EventBase` as it was originally seen over federation (no metadata added). Previously, we returned the same `event` reference that we stored in the cache which downstream code modified in place and added metadata like setting it as an `outlier`  and essentially poisoned our cache. Now we always return a copy of the `event` so the original can stay pristine in our cache and re-used for the next cache call.

Split out from https://github.com/matrix-org/synapse/pull/13205

As discussed at:

 - https://github.com/matrix-org/synapse/pull/13205#discussion_r918365746
 - https://github.com/matrix-org/synapse/pull/13205#discussion_r918366125

Related to https://github.com/matrix-org/synapse/issues/12584. This PR doesn't fix that issue because it hits [`get_event` which exists from the local database before it tries to `get_pdu`](https://github.com/matrix-org/synapse/blob/7864f33e286dec22368dc0b11c06eebb1462a51e/synapse/federation/federation_client.py#L581-L594).

1 files changed, 113 insertions, 12 deletions

diff --git a/tests/federation/test_federation_client.py b/tests/federation/test_federation_client.py
index cf6b130e4f..50e376f695 100644
--- a/tests/federation/test_federation_client.py
+++ b/tests/federation/test_federation_client.py
@@ -22,6 +22,7 @@ from twisted.python.failure import Failure
 from twisted.test.proto_helpers import MemoryReactor
 
 from synapse.api.room_versions import RoomVersions
+from synapse.events import EventBase
 from synapse.server import HomeServer
 from synapse.types import JsonDict
 from synapse.util import Clock
@@ -38,20 +39,24 @@ class FederationClientTest(FederatingHomeserverTestCase):
         self._mock_agent = mock.create_autospec(twisted.web.client.Agent, spec_set=True)
         homeserver.get_federation_http_client().agent = self._mock_agent
 
-    def test_get_room_state(self):
-        creator = f"@creator:{self.OTHER_SERVER_NAME}"
-        test_room_id = "!room_id"
+        # Move clock up to somewhat realistic time so the PDU destination retry
+        # works (`now` needs to be larger than `0 + PDU_RETRY_TIME_MS`).
+        self.reactor.advance(1000000000)
+
+        self.creator = f"@creator:{self.OTHER_SERVER_NAME}"
+        self.test_room_id = "!room_id"
 
+    def test_get_room_state(self):
         # mock up some events to use in the response.
         # In real life, these would have things in `prev_events` and `auth_events`, but that's
         # a bit annoying to mock up, and the code under test doesn't care, so we don't bother.
         create_event_dict = self.add_hashes_and_signatures_from_other_server(
             {
-                "room_id": test_room_id,
+                "room_id": self.test_room_id,
                 "type": "m.room.create",
                 "state_key": "",
-                "sender": creator,
-                "content": {"creator": creator},
+                "sender": self.creator,
+                "content": {"creator": self.creator},
                 "prev_events": [],
                 "auth_events": [],
                 "origin_server_ts": 500,
@@ -59,10 +64,10 @@ class FederationClientTest(FederatingHomeserverTestCase):
         )
         member_event_dict = self.add_hashes_and_signatures_from_other_server(
             {
-                "room_id": test_room_id,
+                "room_id": self.test_room_id,
                 "type": "m.room.member",
-                "sender": creator,
-                "state_key": creator,
+                "sender": self.creator,
+                "state_key": self.creator,
                 "content": {"membership": "join"},
                 "prev_events": [],
                 "auth_events": [],
@@ -71,9 +76,9 @@ class FederationClientTest(FederatingHomeserverTestCase):
         )
         pl_event_dict = self.add_hashes_and_signatures_from_other_server(
             {
-                "room_id": test_room_id,
+                "room_id": self.test_room_id,
                 "type": "m.room.power_levels",
-                "sender": creator,
+                "sender": self.creator,
                 "state_key": "",
                 "content": {},
                 "prev_events": [],
@@ -103,7 +108,7 @@ class FederationClientTest(FederatingHomeserverTestCase):
         state_resp, auth_resp = self.get_success(
             self.hs.get_federation_client().get_room_state(
                 "yet.another.server",
-                test_room_id,
+                self.test_room_id,
                 "event_id",
                 RoomVersions.V9,
             )
@@ -130,6 +135,102 @@ class FederationClientTest(FederatingHomeserverTestCase):
             ["m.room.create", "m.room.member", "m.room.power_levels"],
         )
 
+    def test_get_pdu_returns_nothing_when_event_does_not_exist(self):
+        """No event should be returned when the event does not exist"""
+        remote_pdu = self.get_success(
+            self.hs.get_federation_client().get_pdu(
+                ["yet.another.server"],
+                "event_should_not_exist",
+                RoomVersions.V9,
+            )
+        )
+        self.assertEqual(remote_pdu, None)
+
+    def test_get_pdu(self):
+        """Test to make sure an event is returned by `get_pdu()`"""
+        self._get_pdu_once()
+
+    def test_get_pdu_event_from_cache_is_pristine(self):
+        """Test that modifications made to events returned by `get_pdu()`
+        do not propagate back to to the internal cache (events returned should
+        be a copy).
+        """
+
+        # Get the PDU in the cache
+        remote_pdu = self._get_pdu_once()
+
+        # Modify the the event reference.
+        # This change should not make it back to the `_get_pdu_cache`.
+        remote_pdu.internal_metadata.outlier = True
+
+        # Get the event again. This time it should read it from cache.
+        remote_pdu2 = self.get_success(
+            self.hs.get_federation_client().get_pdu(
+                ["yet.another.server"],
+                remote_pdu.event_id,
+                RoomVersions.V9,
+            )
+        )
+
+        # Sanity check that we are working against the same event
+        self.assertEqual(remote_pdu.event_id, remote_pdu2.event_id)
+
+        # Make sure the event does not include modification from earlier
+        self.assertIsNotNone(remote_pdu2)
+        self.assertEqual(remote_pdu2.internal_metadata.outlier, False)
+
+    def _get_pdu_once(self) -> EventBase:
+        """Retrieve an event via `get_pdu()` and assert that an event was returned.
+        Also used to prime the cache for subsequent test logic.
+        """
+        message_event_dict = self.add_hashes_and_signatures_from_other_server(
+            {
+                "room_id": self.test_room_id,
+                "type": "m.room.message",
+                "sender": self.creator,
+                "state_key": "",
+                "content": {},
+                "prev_events": [],
+                "auth_events": [],
+                "origin_server_ts": 700,
+                "depth": 10,
+            }
+        )
+
+        # mock up the response, and have the agent return it
+        self._mock_agent.request.side_effect = lambda *args, **kwargs: defer.succeed(
+            _mock_response(
+                {
+                    "origin": "yet.another.server",
+                    "origin_server_ts": 900,
+                    "pdus": [
+                        message_event_dict,
+                    ],
+                }
+            )
+        )
+
+        remote_pdu = self.get_success(
+            self.hs.get_federation_client().get_pdu(
+                ["yet.another.server"],
+                "event_id",
+                RoomVersions.V9,
+            )
+        )
+
+        # check the right call got made to the agent
+        self._mock_agent.request.assert_called_once_with(
+            b"GET",
+            b"matrix://yet.another.server/_matrix/federation/v1/event/event_id",
+            headers=mock.ANY,
+            bodyProducer=None,
+        )
+
+        self.assertIsNotNone(remote_pdu)
+        self.assertEqual(remote_pdu.internal_metadata.outlier, False)
+
+        return remote_pdu
+
 
 def _mock_response(resp: JsonDict):
     body = json.dumps(resp).encode("utf-8")