Try to cap incremental syncs from working on massive gaps

matrix-org · May 14, 2024 · a7953fe · a7953fe
1 parent 46902e5
commit a7953fe
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 11 deletions.
diff --git a/syncapi/storage/shared/storage_consumer.go b/syncapi/storage/shared/storage_consumer.go
@@ -508,7 +508,7 @@ func (d *Database) CleanSendToDeviceUpdates(
 // getMembershipFromEvent returns the value of content.membership iff the event is a state event
 // with type 'm.room.member' and state_key of userID. Otherwise, an empty string is returned.
 func getMembershipFromEvent(ctx context.Context, ev gomatrixserverlib.PDU, userID string, rsAPI api.SyncRoomserverAPI) (string, string) {
-	if ev.StateKey() == nil || *ev.StateKey() == "" {
+	if ev.Type() != "m.room.member" || ev.StateKey() == nil || *ev.StateKey() == "" {
 		return "", ""
 	}
 	fullUser, err := spec.NewUserID(userID, true)
@@ -520,7 +520,7 @@ func getMembershipFromEvent(ctx context.Context, ev gomatrixserverlib.PDU, userI
 		return "", ""
 	}
 
-	if ev.Type() != "m.room.member" || !ev.StateKeyEquals(string(*senderID)) {
+	if !ev.StateKeyEquals(string(*senderID)) {
 		return "", ""
 	}
 	membership, err := ev.Membership()

diff --git a/syncapi/sync/requestpool.go b/syncapi/sync/requestpool.go
@@ -397,13 +397,16 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 			}
 		} else {
 			// Incremental sync
+			reasonablePositions := findReasonableIncrementalSyncWindow(
+				syncReq.Since, rp.Notifier.CurrentPosition(),
+			)
 			syncReq.Response.NextBatch = types.StreamingToken{
 				PDUPosition: withTransaction(
 					syncReq.Since.PDUPosition,
 					func(txn storage.DatabaseTransaction) types.StreamPosition {
 						return rp.streams.PDUStreamProvider.IncrementalSync(
 							syncReq.Context, txn, syncReq,
-							syncReq.Since.PDUPosition, rp.Notifier.CurrentPosition().PDUPosition,
+							syncReq.Since.PDUPosition, reasonablePositions.PDUPosition,
 						)
 					},
 				),
@@ -412,7 +415,7 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 					func(txn storage.DatabaseTransaction) types.StreamPosition {
 						return rp.streams.TypingStreamProvider.IncrementalSync(
 							syncReq.Context, txn, syncReq,
-							syncReq.Since.TypingPosition, rp.Notifier.CurrentPosition().TypingPosition,
+							syncReq.Since.TypingPosition, reasonablePositions.TypingPosition,
 						)
 					},
 				),
@@ -421,7 +424,7 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 					func(txn storage.DatabaseTransaction) types.StreamPosition {
 						return rp.streams.ReceiptStreamProvider.IncrementalSync(
 							syncReq.Context, txn, syncReq,
-							syncReq.Since.ReceiptPosition, rp.Notifier.CurrentPosition().ReceiptPosition,
+							syncReq.Since.ReceiptPosition, reasonablePositions.ReceiptPosition,
 						)
 					},
 				),
@@ -430,7 +433,7 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 					func(txn storage.DatabaseTransaction) types.StreamPosition {
 						return rp.streams.InviteStreamProvider.IncrementalSync(
 							syncReq.Context, txn, syncReq,
-							syncReq.Since.InvitePosition, rp.Notifier.CurrentPosition().InvitePosition,
+							syncReq.Since.InvitePosition, reasonablePositions.InvitePosition,
 						)
 					},
 				),
@@ -439,7 +442,7 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 					func(txn storage.DatabaseTransaction) types.StreamPosition {
 						return rp.streams.SendToDeviceStreamProvider.IncrementalSync(
 							syncReq.Context, txn, syncReq,
-							syncReq.Since.SendToDevicePosition, rp.Notifier.CurrentPosition().SendToDevicePosition,
+							syncReq.Since.SendToDevicePosition, reasonablePositions.SendToDevicePosition,
 						)
 					},
 				),
@@ -448,7 +451,7 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 					func(txn storage.DatabaseTransaction) types.StreamPosition {
 						return rp.streams.AccountDataStreamProvider.IncrementalSync(
 							syncReq.Context, txn, syncReq,
-							syncReq.Since.AccountDataPosition, rp.Notifier.CurrentPosition().AccountDataPosition,
+							syncReq.Since.AccountDataPosition, reasonablePositions.AccountDataPosition,
 						)
 					},
 				),
@@ -457,7 +460,7 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 					func(txn storage.DatabaseTransaction) types.StreamPosition {
 						return rp.streams.NotificationDataStreamProvider.IncrementalSync(
 							syncReq.Context, txn, syncReq,
-							syncReq.Since.NotificationDataPosition, rp.Notifier.CurrentPosition().NotificationDataPosition,
+							syncReq.Since.NotificationDataPosition, reasonablePositions.NotificationDataPosition,
 						)
 					},
 				),
@@ -466,7 +469,7 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 					func(txn storage.DatabaseTransaction) types.StreamPosition {
 						return rp.streams.DeviceListStreamProvider.IncrementalSync(
 							syncReq.Context, txn, syncReq,
-							syncReq.Since.DeviceListPosition, rp.Notifier.CurrentPosition().DeviceListPosition,
+							syncReq.Since.DeviceListPosition, reasonablePositions.DeviceListPosition,
 						)
 					},
 				),
@@ -475,7 +478,7 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 					func(txn storage.DatabaseTransaction) types.StreamPosition {
 						return rp.streams.PresenceStreamProvider.IncrementalSync(
 							syncReq.Context, txn, syncReq,
-							syncReq.Since.PresencePosition, rp.Notifier.CurrentPosition().PresencePosition,
+							syncReq.Since.PresencePosition, reasonablePositions.PresencePosition,
 						)
 					},
 				),
@@ -585,3 +588,23 @@ func (rp *RequestPool) shouldReturnImmediately(syncReq *types.SyncRequest, curre
 	}
 	return false
 }
+
+func findReasonableIncrementalSyncWindow(since, limit types.StreamingToken) types.StreamingToken {
+	const windowSize = 100 // TODO: reasonable number?
+	for s, l := range map[*types.StreamPosition]types.StreamPosition{
+		&since.AccountDataPosition:      limit.AccountDataPosition,
+		&since.DeviceListPosition:       limit.DeviceListPosition,
+		&since.InvitePosition:           limit.InvitePosition,
+		&since.NotificationDataPosition: limit.NotificationDataPosition,
+		&since.PDUPosition:              limit.PDUPosition,
+		&since.PresencePosition:         limit.PresencePosition,
+		&since.ReceiptPosition:          limit.ReceiptPosition,
+		&since.SendToDevicePosition:     limit.SendToDevicePosition,
+		&since.TypingPosition:           limit.TypingPosition,
+	} {
+		if *s += windowSize; *s > l {
+			*s = l
+		}
+	}
+	return since
+}