diff --git a/syncapi/streams/stream_invite.go b/syncapi/streams/stream_invite.go index 700f25c10..542f55722 100644 --- a/syncapi/streams/stream_invite.go +++ b/syncapi/streams/stream_invite.go @@ -80,6 +80,7 @@ func (p *InviteStreamProvider) IncrementalSync( if _, ok := req.Response.Rooms.Join[roomID]; ok { continue } + lr := types.NewLeaveResponse() h := sha256.Sum256(append([]byte(roomID), []byte(strconv.FormatInt(int64(to), 10))...)) lr.Timeline.Events = append(lr.Timeline.Events, gomatrixserverlib.ClientEvent{ diff --git a/syncapi/streams/streams.go b/syncapi/streams/streams.go index caa428827..c3b2ac348 100644 --- a/syncapi/streams/streams.go +++ b/syncapi/streams/streams.go @@ -128,7 +128,6 @@ func ToToken(provider StreamProvider, position types.StreamPosition) types.Strea default: panic(fmt.Sprintf("unknown stream provider: %T", t)) } - return types.StreamingToken{} } func IncrementalPositions(provider StreamProvider, current, since types.StreamingToken) (types.StreamPosition, types.StreamPosition) { @@ -154,5 +153,4 @@ func IncrementalPositions(provider StreamProvider, current, since types.Streamin default: panic(fmt.Sprintf("unknown stream provider: %T", t)) } - return 0, 0 } diff --git a/syncapi/sync/request.go b/syncapi/sync/request.go index 620dfdcdb..c54f6db4e 100644 --- a/syncapi/sync/request.go +++ b/syncapi/sync/request.go @@ -21,6 +21,7 @@ import ( "math" "net/http" "strconv" + "sync" "time" "github.com/matrix-org/gomatrixserverlib" @@ -101,6 +102,7 @@ func newSyncRequest(req *http.Request, device userapi.Device, syncDB storage.Dat Rooms: make(map[string]string), // Populated by the PDU stream WantFullState: wantFullState, // MembershipChanges: make(map[string]struct{}), // Populated by the PDU stream + SyncMu: &sync.Mutex{}, }, nil } diff --git a/syncapi/sync/requestpool.go b/syncapi/sync/requestpool.go index 766c58737..2affd1074 100644 --- a/syncapi/sync/requestpool.go +++ b/syncapi/sync/requestpool.go @@ -326,6 +326,8 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi. succeeded = err == nil sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err) }() + syncReq.SyncMu.Lock() + defer syncReq.SyncMu.Unlock() return f(snapshot) } diff --git a/syncapi/types/provider.go b/syncapi/types/provider.go index 9a533002b..614d99273 100644 --- a/syncapi/types/provider.go +++ b/syncapi/types/provider.go @@ -2,6 +2,7 @@ package types import ( "context" + "sync" "time" "github.com/matrix-org/gomatrixserverlib" @@ -26,6 +27,7 @@ type SyncRequest struct { MembershipChanges map[string]struct{} // Updated by the PDU stream. IgnoredUsers IgnoredUsers + SyncMu *sync.Mutex } func (r *SyncRequest) IsRoomPresent(roomID string) bool {