MirrorHub/dendrite
0
0
Fork 0
mirror of https://github.com/matrix-org/dendrite synced 2024-06-13 18:08:59 +02:00
Code Issues Releases Wiki Activity

Merge fb5e1abe71 into 46902e5766

Browse source
This commit is contained in:
Till 2024-05-15 09:03:58 +00:00 committed by GitHub
parent 46902e5766 fb5e1abe71
commit 1aa6913e0e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 37 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
syncapi/storage/shared/storage_consumer.go
View file

@ -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()

44
syncapi/sync/requestpool.go
View file

@ -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
}