diff --git a/syncapi/storage/shared/storage_consumer.go b/syncapi/storage/shared/storage_consumer.go index 923ead9bdd..751bc6ce7b 100644 --- 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 index 5a92c70e1c..448b751ce9 100644 --- a/syncapi/sync/requestpool.go +++ b/syncapi/sync/requestpool.go @@ -397,13 +397,19 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi. } } else { // Incremental sync + reasonablePositions := findReasonableIncrementalSyncWindow( + syncReq.Since, rp.Notifier.CurrentPosition(), + ) + // Also update the currentPos, which is used for the retry logic below. + // Otherwise we may skip over some events. + currentPos = reasonablePositions 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 +418,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 +427,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 +436,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 +445,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 +454,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 +463,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 +472,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 +481,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 +591,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 +}