diff --git a/changelog.d/16973.bugfix b/changelog.d/16973.bugfix
new file mode 100644
index 00000000000..a0e3a123049
--- /dev/null
+++ b/changelog.d/16973.bugfix
@@ -0,0 +1 @@
+Fix joining remote rooms when a module uses the `on_new_event` callback. This callback may now pass partial state events instead of the full state for remote rooms.
diff --git a/docs/modules/third_party_rules_callbacks.md b/docs/modules/third_party_rules_callbacks.md
index d06bff25eb1..b97e28db110 100644
--- a/docs/modules/third_party_rules_callbacks.md
+++ b/docs/modules/third_party_rules_callbacks.md
@@ -142,6 +142,10 @@ Called after sending an event into a room. The module is passed the event, as we
 as the state of the room _after_ the event. This means that if the event is a state event,
 it will be included in this state.
 
+The state map may not be complete if Synapse hasn't yet loaded the full state
+of the room. This can happen for events in rooms that were just joined from
+a remote server.
+
 Note that this callback is called when the event has already been processed and stored
 into the room, which means this callback cannot be used to deny persisting the event. To
 deny an incoming event, see [`check_event_for_spam`](spam_checker_callbacks.md#check_event_for_spam) instead.
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 7a3255aeef6..9f7a04372de 100644
--- a/synapse/module_api/callbacks/third_party_event_rules_callbacks.py
+++ b/synapse/module_api/callbacks/third_party_event_rules_callbacks.py
@@ -366,7 +366,7 @@ async def check_threepid_can_be_invited(
         if len(self._check_threepid_can_be_invited_callbacks) == 0:
             return True
 
-        state_events = await self._get_state_map_for_room(room_id)
+        state_events = await self._storage_controllers.state.get_current_state(room_id)
 
         for callback in self._check_threepid_can_be_invited_callbacks:
             try:
@@ -399,7 +399,7 @@ async def check_visibility_can_be_modified(
         if len(self._check_visibility_can_be_modified_callbacks) == 0:
             return True
 
-        state_events = await self._get_state_map_for_room(room_id)
+        state_events = await self._storage_controllers.state.get_current_state(room_id)
 
         for callback in self._check_visibility_can_be_modified_callbacks:
             try:
@@ -427,7 +427,13 @@ async def on_new_event(self, event_id: str) -> None:
             return
 
         event = await self.store.get_event(event_id)
-        state_events = await self._get_state_map_for_room(event.room_id)
+
+        # 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.
+        # This would end up in a deadlock.
+        state_events = await self._storage_controllers.state.get_current_state(
+            event.room_id, await_full_state=False
+        )
 
         for callback in self._on_new_event_callbacks:
             try:
@@ -490,17 +496,6 @@ async def check_can_deactivate_user(
                 )
         return True
 
-    async def _get_state_map_for_room(self, room_id: str) -> StateMap[EventBase]:
-        """Given a room ID, return the state events of that room.
-
-        Args:
-            room_id: The ID of the room.
-
-        Returns:
-            A dict mapping (event type, state key) to state event.
-        """
-        return await self._storage_controllers.state.get_current_state(room_id)
-
     async def on_profile_update(
         self, user_id: str, new_profile: ProfileInfo, by_admin: bool, deactivation: bool
     ) -> None:
diff --git a/synapse/storage/controllers/state.py b/synapse/storage/controllers/state.py
index 99666d79a9b..22d93a561c6 100644
--- a/synapse/storage/controllers/state.py
+++ b/synapse/storage/controllers/state.py
@@ -562,10 +562,15 @@ async def get_current_state_deltas(
     @trace
     @tag_args
     async def get_current_state(
-        self, room_id: str, state_filter: Optional[StateFilter] = None
+        self,
+        room_id: str,
+        state_filter: Optional[StateFilter] = None,
+        await_full_state: bool = True,
     ) -> StateMap[EventBase]:
         """Same as `get_current_state_ids` but also fetches the events"""
-        state_map_ids = await self.get_current_state_ids(room_id, state_filter)
+        state_map_ids = await self.get_current_state_ids(
+            room_id, state_filter, await_full_state
+        )
 
         event_map = await self.stores.main.get_events(list(state_map_ids.values()))