Skip on_new_event module callbacks for rejected events

`ThirdPartyEventRulesModuleApiCallbacks.on_new_event` looked the event up with
`store.get_event(event_id)`, which raises `NotFoundError` for a rejected event.
The notifier calls it for every new event, so on any deployment with a module
registering an `on_new_event` callback, a rejected event received over
federation failed the persist that triggered it.

The lookup now passes `allow_rejected=True` and returns early on a rejected
event. `docs/modules/third_party_rules_callbacks.md` has promised this since
2023: "This callback will not be called for events that are marked as
rejected."

Only the instance that persisted the event could hit this: `ReplicationDataHandler`
already skips rejected rows before notifying, so a worker reading the events
stream never got there.

The test registers a callback, sends an event that will be rejected, and checks
that the persist succeeds and that the callback is not told about the event. It
fails with `NotFoundError` without the fix.

Split out of #19958.

Co-authored-by: Quentin Gliech <quenting@element.io>
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

changelog.d/20442.bugfix

@@ -0,0 +1 @@
+Fix event persistence failing when a rejected event arrives over federation and a module has registered an `on_new_event` callback. Such callbacks are now skipped for rejected events, as the module documentation says they are.

synapse/module_api/callbacks/third_party_event_rules_callbacks.py

@@ -416,7 +416,12 @@ class ThirdPartyEventRulesModuleApiCallbacks:
         if len(self._on_new_event_callbacks) == 0:
             return
 
-        event = await self.store.get_event(event_id)
+        # Rejected events are not shown to modules. Without `allow_rejected` a
+        # rejected event raises NotFoundError here, which propagates out of the
+        # notifier into the caller persisting further federation events.
+        event = await self.store.get_event(event_id, allow_rejected=True)
+        if event.rejected_reason is not None:
+            return
 
         # We *don't* want to wait for the full state here, because waiting for full
         # state will persist event, which in turn will call this method.

tests/handlers/test_federation.py

@@ -199,6 +199,61 @@ class FederationTestCase(unittest.FederatingHomeserverTestCase):
 
         self.assertEqual(sg, sg2)
 
+    def test_rejected_event_with_on_new_event_callback(self) -> None:
+        """
+        A rejected event must not break the notifier when a module has
+        registered an `on_new_event` callback (which the notifier invokes for
+        each new event), and the callback must not be told about it.
+
+        Without `allow_rejected=True` in the dispatcher, the event lookup
+        raises `NotFoundError` inside the federation persist path.
+        """
+        OTHER_SERVER = "otherserver"
+        OTHER_USER = "@otheruser:" + OTHER_SERVER
+
+        # create the room
+        user_id = self.register_user("kermit", "test")
+        tok = self.login("kermit", "test")
+        room_id = self.helper.create_room_as(room_creator=user_id, tok=tok)
+        room_version = self.get_success(self.store.get_room_version(room_id))
+
+        # pretend that another server has joined
+        join_event = self._build_and_send_join_event(OTHER_SERVER, OTHER_USER, room_id)
+
+        # register an `on_new_event` callback (after the events above, so that
+        # they don't count towards its calls)
+        on_new_event = AsyncMock(return_value=None)
+        self.hs.get_module_api().register_third_party_rules_callbacks(
+            on_new_event=on_new_event
+        )
+
+        # build and send an event which will be rejected
+        ev = make_test_pdu_event(
+            {
+                "type": EventTypes.Message,
+                "content": {},
+                "room_id": room_id,
+                "sender": "@yetanotheruser:" + OTHER_SERVER,
+                "depth": join_event.depth + 1,
+                "prev_events": [join_event.event_id],
+                "auth_events": [],
+                "origin_server_ts": self.clock.time_msec(),
+            },
+            room_version,
+        )
+
+        # this should succeed, despite the on_new_event callback
+        self.get_success(
+            self.hs.get_federation_event_handler().on_receive_pdu(OTHER_SERVER, ev)
+        )
+
+        # that should have been rejected...
+        e = self.get_success(self.store.get_event(ev.event_id, allow_rejected=True))
+        self.assertIsNotNone(e.rejected_reason)
+
+        # ... and the module should not have been told about it
+        on_new_event.assert_not_called()
+
     def test_rejected_state_event_state(self) -> None:
         """
         Check that we store the state group correctly for rejected state events.