@log_function
    @defer.inlineCallbacks
    def on_receive_pdu(self, origin, pdu, state=None,
                       auth_chain=None):
        event = pdu

        logger.debug("Got event: %s", event.event_id)

        # If we are currently in the process of joining this room, then we
        # queue up events for later processing.
        if event.room_id in self.room_queues:
            self.room_queues[event.room_id].append((pdu, origin))
            return

        logger.debug("Processing event: %s", event.event_id)

        logger.debug("Event: %s", event)

        # FIXME (erikj): Awful hack to make the case where we are not currently
        # in the room work
        is_in_room = yield self.auth.check_host_in_room(
            event.room_id,
            self.server_name
        )
        if not is_in_room and not event.internal_metadata.is_outlier():
            logger.debug("Got event for room we're not in.")

            try:
                event_stream_id, max_stream_id = yield self._persist_auth_tree(
                    auth_chain, state, event
                )
            except AuthError as e:
                raise FederationError(
                    "ERROR",
                    e.code,
                    e.msg,
                    affected=event.event_id,
                )

        else:
            event_ids = set()
            if state:
                event_ids |= {e.event_id for e in state}
            if auth_chain:
                event_ids |= {e.event_id for e in auth_chain}

            seen_ids = set(
                (yield self.store.have_events(event_ids)).keys()
            )

            if state and auth_chain is not None:
                # If we have any state or auth_chain given to us by the replication
                # layer, then we should handle them (if we haven't before.)

                event_infos = []

                for e in itertools.chain(auth_chain, state):
                    if e.event_id in seen_ids:
                        continue
                    e.internal_metadata.outlier = True
                    auth_ids = [e_id for e_id, _ in e.auth_events]
                    auth = {
                        (e.type, e.state_key): e for e in auth_chain
                        if e.event_id in auth_ids or e.type == EventTypes.Create
                    }
                    event_infos.append({
                        "event": e,
                        "auth_events": auth,
                    })
                    seen_ids.add(e.event_id)

                yield self._handle_new_events(
                    origin,
                    event_infos,
                    outliers=True
                )

            try:
                context, event_stream_id, max_stream_id = yield self._handle_new_event(
                    origin,
                    event,
                    state=state,
                )
            except AuthError as e:
                raise FederationError(
                    "ERROR",
                    e.code,
                    e.msg,
                    affected=event.event_id,
                )

        # if we're receiving valid events from an origin,
        # it's probably a good idea to mark it as not in retry-state
        # for sending (although this is a bit of a leap)
        retry_timings = yield self.store.get_destination_retry_timings(origin)
        if retry_timings and retry_timings["retry_last_ts"]:
            self.store.set_destination_retry_timings(origin, 0, 0)

        room = yield self.store.get_room(event.room_id)

        if not room:
            try:
                yield self.store.store_room(
                    room_id=event.room_id,
                    room_creator_user_id="",
                    is_public=False,
                )
            except StoreError:
                logger.exception("Failed to store room.")

        extra_users = []
        if event.type == EventTypes.Member:
            target_user_id = event.state_key
            target_user = UserID.from_string(target_user_id)
            extra_users.append(target_user)

        with PreserveLoggingContext():
            self.notifier.on_new_room_event(
                event, event_stream_id, max_stream_id,
                extra_users=extra_users
            )

        if event.type == EventTypes.Member:
            if event.membership == Membership.JOIN:
                prev_state = context.current_state.get((event.type, event.state_key))
                if not prev_state or prev_state.membership != Membership.JOIN:
                    # Only fire user_joined_room if the user has acutally
                    # joined the room. Don't bother if the user is just
                    # changing their profile info.
                    user = UserID.from_string(event.state_key)
                    yield user_joined_room(self.distributor, user, event.room_id)
Called by the ReplicationLayer when we have a new pdu. We need to
dataset/ETHPy150Open matrix-org/synapse/synapse/handlers/federation.py/FederationHandler.on_receive_pdu