diff --git a/changelog.d/20442.bugfix b/changelog.d/20442.bugfix
new file mode 100644
index 0000000000..8772520ab2
--- /dev/null
+++ b/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.
diff --git a/synapse/module_api/callbacks/third_party_event_rules_callbacks.py b/synapse/module_api/callbacks/third_party_event_rules_callbacks.py
index 68053d65a3..d19759ab44 100644
--- a/synapse/module_api/callbacks/third_party_event_rules_callbacks.py
+++ b/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.
diff --git a/tests/handlers/test_federation.py b/tests/handlers/test_federation.py
index edb419744a..784073bfd5 100644
--- a/tests/handlers/test_federation.py
+++ b/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.