Update onOldRoomEvent

syncapi/consumers/roomserver.go

@@ -207,10 +207,10 @@ func (s *OutputRoomEventConsumer) onOldRoomEvent(
 		ctx,
 		ev,
 		[]*gomatrixserverlib.HeaderedEvent{},
-		[]string{},           // adds no state
+		[]string{}, // adds no state
-		[]string{},           // removes no state
+		[]string{}, // removes no state
-		nil,                  // no transaction
+		nil,        // no transaction
-		ev.StateKey() != nil, // exclude from sync?
+		true,       // exclude from sync?
 	)
 	if err != nil {
 		// panic rather than continue with an inconsistent database
@@ -221,13 +221,8 @@ func (s *OutputRoomEventConsumer) onOldRoomEvent(
 		return nil
 	}
 
-	if pduPos, err = s.notifyJoinedPeeks(ctx, ev, pduPos); err != nil {
-		log.WithError(err).Errorf("Failed to notifyJoinedPeeks for PDU pos %d", pduPos)
-		return err
-	}
-
 	s.pduStream.Advance(pduPos)
-	s.notifier.OnNewEvent(ev, ev.RoomID(), nil, types.StreamingToken{PDUPosition: pduPos})
+	s.notifier.OnOldEvent(ev, ev.RoomID(), nil, types.StreamingToken{PDUPosition: pduPos})
 
 	return nil
 }

syncapi/notifier/notifier.go

@@ -124,6 +124,21 @@ func (n *Notifier) OnNewEvent(
 	}
 }
 
+// OnOldEvent is called when the sync API receives an "old" event, e.g.
+// because it was pulled in by get_missing_events or similar. We need
+// to update the stream position, but we don't bother waking up any
+// clients because the event will be stored as excluded from sync anyway.
+func (n *Notifier) OnOldEvent(
+	ev *gomatrixserverlib.HeaderedEvent, roomID string, userIDs []string,
+	posUpdate types.StreamingToken,
+) {
+	n.streamLock.Lock()
+	defer n.streamLock.Unlock()
+
+	n.currPos.ApplyUpdates(posUpdate)
+	n.removeEmptyUserStreams()
+}
+
 func (n *Notifier) OnNewAccountData(
 	userID string, posUpdate types.StreamingToken,
 ) {