diff options
Diffstat (limited to 'syncapi/sync/requestpool.go')
| -rw-r--r-- | syncapi/sync/requestpool.go | 53 |
1 files changed, 33 insertions, 20 deletions
diff --git a/syncapi/sync/requestpool.go b/syncapi/sync/requestpool.go index b2ea105f..1d0ac1a4 100644 --- a/syncapi/sync/requestpool.go +++ b/syncapi/sync/requestpool.go @@ -305,6 +305,13 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi. syncReq.Log.WithField("currentPos", currentPos).Debugln("Responding to sync immediately") } + snapshot, err := rp.db.NewDatabaseSnapshot(req.Context()) + if err != nil { + logrus.WithError(err).Error("Failed to acquire database snapshot for sync request") + return jsonerror.InternalServerError() + } + defer snapshot.Rollback() // nolint:errcheck + if syncReq.Since.IsEmpty() { // Complete sync syncReq.Response.NextBatch = types.StreamingToken{ @@ -312,70 +319,70 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi. // might advance while processing other streams, resulting in flakey // tests. DeviceListPosition: rp.streams.DeviceListStreamProvider.CompleteSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, ), PDUPosition: rp.streams.PDUStreamProvider.CompleteSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, ), TypingPosition: rp.streams.TypingStreamProvider.CompleteSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, ), ReceiptPosition: rp.streams.ReceiptStreamProvider.CompleteSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, ), InvitePosition: rp.streams.InviteStreamProvider.CompleteSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, ), SendToDevicePosition: rp.streams.SendToDeviceStreamProvider.CompleteSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, ), AccountDataPosition: rp.streams.AccountDataStreamProvider.CompleteSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, ), NotificationDataPosition: rp.streams.NotificationDataStreamProvider.CompleteSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, ), PresencePosition: rp.streams.PresenceStreamProvider.CompleteSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, ), } } else { // Incremental sync syncReq.Response.NextBatch = types.StreamingToken{ PDUPosition: rp.streams.PDUStreamProvider.IncrementalSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, syncReq.Since.PDUPosition, currentPos.PDUPosition, ), TypingPosition: rp.streams.TypingStreamProvider.IncrementalSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, syncReq.Since.TypingPosition, currentPos.TypingPosition, ), ReceiptPosition: rp.streams.ReceiptStreamProvider.IncrementalSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, syncReq.Since.ReceiptPosition, currentPos.ReceiptPosition, ), InvitePosition: rp.streams.InviteStreamProvider.IncrementalSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, syncReq.Since.InvitePosition, currentPos.InvitePosition, ), SendToDevicePosition: rp.streams.SendToDeviceStreamProvider.IncrementalSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, syncReq.Since.SendToDevicePosition, currentPos.SendToDevicePosition, ), AccountDataPosition: rp.streams.AccountDataStreamProvider.IncrementalSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, syncReq.Since.AccountDataPosition, currentPos.AccountDataPosition, ), NotificationDataPosition: rp.streams.NotificationDataStreamProvider.IncrementalSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, syncReq.Since.NotificationDataPosition, currentPos.NotificationDataPosition, ), DeviceListPosition: rp.streams.DeviceListStreamProvider.IncrementalSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, syncReq.Since.DeviceListPosition, currentPos.DeviceListPosition, ), PresencePosition: rp.streams.PresenceStreamProvider.IncrementalSync( - syncReq.Context, syncReq, + syncReq.Context, snapshot, syncReq, syncReq.Since.PresencePosition, currentPos.PresencePosition, ), } @@ -437,9 +444,15 @@ func (rp *RequestPool) OnIncomingKeyChangeRequest(req *http.Request, device *use util.GetLogger(req.Context()).WithError(err).Error("newSyncRequest failed") return jsonerror.InternalServerError() } - rp.streams.PDUStreamProvider.IncrementalSync(req.Context(), syncReq, fromToken.PDUPosition, toToken.PDUPosition) + snapshot, err := rp.db.NewDatabaseSnapshot(req.Context()) + if err != nil { + logrus.WithError(err).Error("Failed to acquire database snapshot for key change") + return jsonerror.InternalServerError() + } + defer snapshot.Rollback() // nolint:errcheck + rp.streams.PDUStreamProvider.IncrementalSync(req.Context(), snapshot, syncReq, fromToken.PDUPosition, toToken.PDUPosition) _, _, err = internal.DeviceListCatchup( - req.Context(), rp.db, rp.keyAPI, rp.rsAPI, syncReq.Device.UserID, + req.Context(), snapshot, rp.keyAPI, rp.rsAPI, syncReq.Device.UserID, syncReq.Response, fromToken.DeviceListPosition, toToken.DeviceListPosition, ) if err != nil { |