sigb.us/dendrite
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases 1 Wiki Activity

Try to cap incremental syncs from working on massive gaps

Browse source
This commit is contained in:
Neil Alexander 2024-04-16 17:49:54 +01:00 committed by Till Faelligen
parent 46902e5766
commit a7953fe3ff
No known key found for this signature in database
GPG key ID: 3DF82D8AB9211D4E
2 changed files with 34 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 // 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. // 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) { 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 "", "" return "", ""
} }
fullUser, err := spec.NewUserID(userID, true) fullUser, err := spec.NewUserID(userID, true)
@ -520,7 +520,7 @@ func getMembershipFromEvent(ctx context.Context, ev gomatrixserverlib.PDU, userI
return "", "" return "", ""
} }
if ev.Type() != "m.room.member" || !ev.StateKeyEquals(string(*senderID)) { if !ev.StateKeyEquals(string(*senderID)) {
return "", "" return "", ""
} }
membership, err := ev.Membership() membership, err := ev.Membership()

41
syncapi/sync/requestpool.go
View file

@ -397,13 +397,16 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
} }
} else { } else {
// Incremental sync // Incremental sync
reasonablePositions := findReasonableIncrementalSyncWindow(
syncReq.Since, rp.Notifier.CurrentPosition(),
)
syncReq.Response.NextBatch = types.StreamingToken{ syncReq.Response.NextBatch = types.StreamingToken{
PDUPosition: withTransaction( PDUPosition: withTransaction(
syncReq.Since.PDUPosition, syncReq.Since.PDUPosition,
func(txn storage.DatabaseTransaction) types.StreamPosition { func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.PDUStreamProvider.IncrementalSync( return rp.streams.PDUStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq, 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 { func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.TypingStreamProvider.IncrementalSync( return rp.streams.TypingStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq, 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 { func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.ReceiptStreamProvider.IncrementalSync( return rp.streams.ReceiptStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq, 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 { func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.InviteStreamProvider.IncrementalSync( return rp.streams.InviteStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq, 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 { func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.SendToDeviceStreamProvider.IncrementalSync( return rp.streams.SendToDeviceStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq, 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 { func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.AccountDataStreamProvider.IncrementalSync( return rp.streams.AccountDataStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq, 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 { func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.NotificationDataStreamProvider.IncrementalSync( return rp.streams.NotificationDataStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq, 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 { func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.DeviceListStreamProvider.IncrementalSync( return rp.streams.DeviceListStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq, 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 { func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.PresenceStreamProvider.IncrementalSync( return rp.streams.PresenceStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq, 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 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
}