diff options
| author | Neil Alexander <neilalexander@users.noreply.github.com> | 2022-09-30 16:34:06 +0100 |
|---|---|---|
| committer | Neil Alexander <neilalexander@users.noreply.github.com> | 2022-09-30 16:34:06 +0100 |
| commit | 7d9545ceead42b4761ee9cce169bc8f5858521ed (patch) | |
| tree | 766c21d14e3df1b6195937c35ee9734daa819ca6 /syncapi | |
| parent | 3617d5a0ff580c7ca0058490308f7f946df3882f (diff) | |
Another `/sync` fix
Diffstat (limited to 'syncapi')
| -rw-r--r-- | syncapi/streams/stream_pdu.go | 9 |
1 files changed, 8 insertions, 1 deletions
diff --git a/syncapi/streams/stream_pdu.go b/syncapi/streams/stream_pdu.go index 43e2ce8b..ba90706e 100644 --- a/syncapi/streams/stream_pdu.go +++ b/syncapi/streams/stream_pdu.go @@ -156,11 +156,13 @@ func (p *PDUStreamProvider) IncrementalSync( if req.WantFullState { if stateDeltas, syncJoinedRooms, err = snapshot.GetStateDeltasForFullStateSync(ctx, req.Device, r, req.Device.UserID, &stateFilter); err != nil { req.Log.WithError(err).Error("p.DB.GetStateDeltasForFullStateSync failed") + _ = snapshot.Rollback() return } } else { if stateDeltas, syncJoinedRooms, err = snapshot.GetStateDeltas(ctx, req.Device, r, req.Device.UserID, &stateFilter); err != nil { req.Log.WithError(err).Error("p.DB.GetStateDeltas failed") + _ = snapshot.Rollback() return } } @@ -175,6 +177,7 @@ func (p *PDUStreamProvider) IncrementalSync( if err = p.addIgnoredUsersToFilter(ctx, snapshot, req, &eventFilter); err != nil { req.Log.WithError(err).Error("unable to update event filter with ignored users") + _ = snapshot.Rollback() } newPos = from @@ -194,7 +197,11 @@ func (p *PDUStreamProvider) IncrementalSync( var pos types.StreamPosition if pos, err = p.addRoomDeltaToResponse(ctx, snapshot, req.Device, newRange, delta, &eventFilter, &stateFilter, req.Response); err != nil { req.Log.WithError(err).Error("d.addRoomDeltaToResponse failed") - return to + _ = snapshot.Rollback() + if err == context.DeadlineExceeded || err == context.Canceled { + return newPos + } + continue // return to } // Reset the position, as it is only for the special case of newly joined rooms if delta.NewlyJoined { |