Don't generate output events for rewrites, but instead notify that state is rewritten on the final new event

roomserver/api/output.go

@@ -86,9 +86,9 @@ const (
 type OutputNewRoomEvent struct {
 	// The Event.
 	Event gomatrixserverlib.HeaderedEvent `json:"event"`
-	// Is the event a timeline event or a state event? Defaults to timeline
+	// Does the event completely rewrite the room state? If so, then AddsStateEventIDs
-	// if not specified.
+	// will contain the entire room state.
-	Type OutputRoomEventType `json:"type"`
+	RewritesState bool `json:"rewrites_state"`
 	// The latest events in the room after this event.
 	// This can be used to set the prev events for new events in the room.
 	// This also can be used to get the full current state after this event.

roomserver/internal/input/input_events.go

@@ -86,7 +86,7 @@ func (r *Inputer) processRoomEvent(
 			"event_id": event.EventID(),
 			"type":     event.Type(),
 			"room":     event.RoomID(),
-		}).Info("Stored outlier")
+		}).Debug("Stored outlier")
 		return event.EventID(), nil
 	}
 
@@ -107,14 +107,23 @@ func (r *Inputer) processRoomEvent(
 		}
 	}
 
+	if input.Kind == api.KindRewrite {
+		logrus.WithFields(logrus.Fields{
+			"event_id": event.EventID(),
+			"type":     event.Type(),
+			"room":     event.RoomID(),
+		}).Debug("Stored rewrite")
+		return event.EventID(), nil
+	}
+
 	if err = r.updateLatestEvents(
-		ctx,                           // context
+		ctx,                 // context
-		roomInfo,                      // room info for the room being updated
+		roomInfo,            // room info for the room being updated
-		stateAtEvent,                  // state at event (below)
+		stateAtEvent,        // state at event (below)
-		event,                         // event
+		event,               // event
-		input.SendAsServer,            // send as server
+		input.SendAsServer,  // send as server
-		input.TransactionID,           // transaction ID
+		input.TransactionID, // transaction ID
-		input.Kind == api.KindRewrite, // historical
+		input.HasState,      // rewrites state?
 	); err != nil {
 		return "", fmt.Errorf("r.updateLatestEvents: %w", err)
 	}

roomserver/internal/input/input_latest_events.go

@@ -54,7 +54,7 @@ func (r *Inputer) updateLatestEvents(
 	event gomatrixserverlib.Event,
 	sendAsServer string,
 	transactionID *api.TransactionID,
-	isHistorical bool,
+	rewritesState bool,
 ) (err error) {
 	updater, err := r.DB.GetLatestEventsForUpdate(ctx, *roomInfo)
 	if err != nil {
@@ -72,7 +72,7 @@ func (r *Inputer) updateLatestEvents(
 		event:         event,
 		sendAsServer:  sendAsServer,
 		transactionID: transactionID,
-		isHistorical:  isHistorical,
+		rewritesState: rewritesState,
 	}
 
 	if err = u.doUpdateLatestEvents(); err != nil {
@@ -95,7 +95,7 @@ type latestEventsUpdater struct {
 	stateAtEvent  types.StateAtEvent
 	event         gomatrixserverlib.Event
 	transactionID *api.TransactionID
-	isHistorical  bool
+	rewritesState bool
 	// Which server to send this event as.
 	sendAsServer string
 	// The eventID of the event that was processed before this one.
@@ -307,14 +307,9 @@ func (u *latestEventsUpdater) makeOutputNewRoomEvent() (*api.OutputEvent, error)
 		latestEventIDs[i] = u.latest[i].EventID
 	}
 
-	var outputType api.OutputRoomEventType
-	if u.isHistorical {
-		outputType = api.OutputRoomState
-	}
-
 	ore := api.OutputNewRoomEvent{
 		Event:           u.event.Headered(u.roomInfo.RoomVersion),
-		Type:            outputType,
+		RewritesState:   u.rewritesState,
 		LastSentEventID: u.lastEventIDSent,
 		LatestEventIDs:  latestEventIDs,
 		TransactionID:   u.transactionID,

syncapi/consumers/roomserver.go

@@ -144,7 +144,7 @@ func (s *OutputRoomEventConsumer) onNewRoomEvent(
 		}
 	}
 
-	if msg.Type == api.OutputRoomState {
+	if msg.RewritesState {
 		err = s.db.RewriteState(
 			ctx,
 			&ev,
@@ -155,7 +155,6 @@ func (s *OutputRoomEventConsumer) onNewRoomEvent(
 		if err != nil {
 			return fmt.Errorf("s.db.RewriteState: %w", err)
 		}
-		return nil
 	}
 
 	pduPos, err := s.db.WriteEvent(

syncapi/routing/messages.go

@@ -417,7 +417,7 @@ func (r *messagesReq) backfill(roomID string, backwardsExtremities map[string][]
 			[]gomatrixserverlib.HeaderedEvent{},
 			[]string{},
 			[]string{},
-			nil, true,
+			nil, false,
 		)
 		if err != nil {
 			return nil, err

syncapi/storage/shared/syncserver.go

@@ -267,12 +267,14 @@ func (d *Database) RewriteState(
 			}
 		}
 
-		if err := d.handleBackwardExtremities(ctx, txn, ev); err != nil {
+		/*
-			return fmt.Errorf("d.handleBackwardExtremities: %w", err)
+			if err := d.handleBackwardExtremities(ctx, txn, ev); err != nil {
-		}
+				return fmt.Errorf("d.handleBackwardExtremities: %w", err)
-
+			}
+		*/
 		// TODO: is there something better here that we can do instead of giving stream position 0?
-		return d.updateRoomState(ctx, txn, []string{}, addStateEvents, types.StreamPosition(0))
+		//return d.updateRoomState(ctx, txn, []string{}, addStateEvents, types.StreamPosition(0))
+		return nil
 	})
 }