Fix broken /sync due to transaction error

syncapi/consumers/roomserver.go

@@ -27,6 +27,7 @@ import (
 	"github.com/tidwall/gjson"
 
 	"github.com/matrix-org/dendrite/internal/fulltext"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/setup/config"
 	"github.com/matrix-org/dendrite/setup/jetstream"
@@ -454,7 +455,8 @@ func (s *OutputRoomEventConsumer) updateStateEvent(event *gomatrixserverlib.Head
 	if err != nil {
 		return nil, err
 	}
-	defer snapshot.Rollback() // nolint:errcheck
+	var succeeded bool
+	defer sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err)
 
 	prevEvent, err := snapshot.GetStateEvent(
 		s.ctx, event.RoomID(), event.Type(), stateKey,
@@ -474,6 +476,7 @@ func (s *OutputRoomEventConsumer) updateStateEvent(event *gomatrixserverlib.Head
 	}
 
 	event.Event, err = event.SetUnsigned(prev)
+	succeeded = true
 	return event, err
 }
 

syncapi/notifier/notifier.go

@@ -19,6 +19,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/syncapi/storage"
 	"github.com/matrix-org/dendrite/syncapi/types"
 	"github.com/matrix-org/gomatrixserverlib"
@@ -323,7 +324,8 @@ func (n *Notifier) Load(ctx context.Context, db storage.Database) error {
 	if err != nil {
 		return err
 	}
-	defer snapshot.Rollback() // nolint:errcheck
+	var succeeded bool
+	defer sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err)
 
 	roomToUsers, err := snapshot.AllJoinedUsersInRooms(ctx)
 	if err != nil {
@@ -337,6 +339,7 @@ func (n *Notifier) Load(ctx context.Context, db storage.Database) error {
 	}
 	n.setPeekingDevices(roomToPeekingDevices)
 
+	succeeded = true
 	return nil
 }
 
@@ -349,7 +352,8 @@ func (n *Notifier) LoadRooms(ctx context.Context, db storage.Database, roomIDs [
 	if err != nil {
 		return err
 	}
-	defer snapshot.Rollback() // nolint:errcheck
+	var succeeded bool
+	defer sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err)
 
 	roomToUsers, err := snapshot.AllJoinedUsersInRoom(ctx, roomIDs)
 	if err != nil {
@@ -357,6 +361,7 @@ func (n *Notifier) LoadRooms(ctx context.Context, db storage.Database, roomIDs [
 	}
 	n.setUsersJoinedToRooms(roomToUsers)
 
+	succeeded = true
 	return nil
 }
 

syncapi/routing/context.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
 	"github.com/matrix-org/dendrite/internal/caching"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
 	roomserver "github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/syncapi/internal"
 	"github.com/matrix-org/dendrite/syncapi/storage"
@@ -55,7 +56,8 @@ func Context(
 	if err != nil {
 		return jsonerror.InternalServerError()
 	}
-	defer snapshot.Rollback() // nolint:errcheck
+	var succeeded bool
+	defer sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err)
 
 	filter, err := parseRoomEventFilter(req)
 	if err != nil {
@@ -184,6 +186,7 @@ func Context(
 		response.End = end.String()
 		response.Start = start.String()
 	}
+	succeeded = true
 	return util.JSONResponse{
 		Code: http.StatusOK,
 		JSON: response,

syncapi/routing/search.go

@@ -31,6 +31,7 @@ import (
 	"github.com/matrix-org/dendrite/clientapi/httputil"
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
 	"github.com/matrix-org/dendrite/internal/fulltext"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/syncapi/storage"
 	"github.com/matrix-org/dendrite/userapi/api"
 )
@@ -65,7 +66,8 @@ func Search(req *http.Request, device *api.Device, syncDB storage.Database, fts
 	if err != nil {
 		return jsonerror.InternalServerError()
 	}
-	defer snapshot.Rollback() // nolint:errcheck
+	var succeeded bool
+	defer sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err)
 
 	// only search rooms the user is actually joined to
 	joinedRooms, err := snapshot.RoomIDsWithMembership(ctx, device.UserID, "join")
@@ -249,6 +251,7 @@ func Search(req *http.Request, device *api.Device, syncDB storage.Database, fts
 
 	logrus.Debugf("Full search request took %v", time.Since(start))
 
+	succeeded = true
 	return util.JSONResponse{
 		Code: http.StatusOK,
 		JSON: res,

syncapi/storage/interface.go

@@ -20,6 +20,7 @@ import (
 	"github.com/matrix-org/gomatrixserverlib"
 
 	"github.com/matrix-org/dendrite/internal/eventutil"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/syncapi/storage/shared"
 	"github.com/matrix-org/dendrite/syncapi/types"
@@ -27,6 +28,7 @@ import (
 )
 
 type DatabaseTransaction interface {
+	sqlutil.Transaction
 	SharedUsers
 
 	MaxStreamPositionForPDUs(ctx context.Context) (types.StreamPosition, error)

syncapi/streams/stream_accountdata.go

@@ -54,6 +54,7 @@ func (p *AccountDataStreamProvider) IncrementalSync(
 	)
 	if err != nil {
 		req.Log.WithError(err).Error("p.DB.GetAccountDataInRange failed")
+		_ = snapshot.Rollback()
 		return from
 	}
 

syncapi/streams/stream_devicelist.go

@@ -34,11 +34,13 @@ func (p *DeviceListStreamProvider) IncrementalSync(
 	to, _, err = internal.DeviceListCatchup(context.Background(), snapshot, p.keyAPI, p.rsAPI, req.Device.UserID, req.Response, from, to)
 	if err != nil {
 		req.Log.WithError(err).Error("internal.DeviceListCatchup failed")
+		_ = snapshot.Rollback()
 		return from
 	}
 	err = internal.DeviceOTKCounts(req.Context, p.keyAPI, req.Device.UserID, req.Device.ID, req.Response)
 	if err != nil {
 		req.Log.WithError(err).Error("internal.DeviceListCatchup failed")
+		_ = snapshot.Rollback()
 		return from
 	}
 

syncapi/streams/stream_invite.go

@@ -56,6 +56,7 @@ func (p *InviteStreamProvider) IncrementalSync(
 	)
 	if err != nil {
 		req.Log.WithError(err).Error("p.DB.InviteEventsInRange failed")
+		_ = snapshot.Rollback()
 		return from
 	}
 

syncapi/streams/stream_notificationdata.go

@@ -46,6 +46,7 @@ func (p *NotificationDataStreamProvider) IncrementalSync(
 	countsByRoom, err := snapshot.GetUserUnreadNotificationCountsForRooms(ctx, req.Device.UserID, req.Rooms)
 	if err != nil {
 		req.Log.WithError(err).Error("GetUserUnreadNotificationCountsForRooms failed")
+		_ = snapshot.Rollback()
 		return from
 	}
 

syncapi/streams/stream_pdu.go

@@ -75,6 +75,7 @@ func (p *PDUStreamProvider) CompleteSync(
 	joinedRoomIDs, err := snapshot.RoomIDsWithMembership(ctx, req.Device.UserID, gomatrixserverlib.Join)
 	if err != nil {
 		req.Log.WithError(err).Error("p.DB.RoomIDsWithMembership failed")
+		_ = snapshot.Rollback()
 		return from
 	}
 
@@ -101,6 +102,7 @@ func (p *PDUStreamProvider) CompleteSync(
 		)
 		if jerr != nil {
 			req.Log.WithError(jerr).Error("p.getJoinResponseForCompleteSync failed")
+			_ = snapshot.Rollback()
 			continue // return from
 		}
 		req.Response.Rooms.Join[roomID] = *jr
@@ -111,6 +113,7 @@ func (p *PDUStreamProvider) CompleteSync(
 	peeks, err := snapshot.PeeksInRange(ctx, req.Device.UserID, req.Device.ID, r)
 	if err != nil {
 		req.Log.WithError(err).Error("p.DB.PeeksInRange failed")
+		_ = snapshot.Rollback()
 		return from
 	}
 	for _, peek := range peeks {
@@ -121,6 +124,7 @@ func (p *PDUStreamProvider) CompleteSync(
 			)
 			if err != nil {
 				req.Log.WithError(err).Error("p.getJoinResponseForCompleteSync failed")
+				_ = snapshot.Rollback()
 				continue // return from
 			}
 			req.Response.Rooms.Peek[peek.RoomID] = *jr

syncapi/streams/stream_presence.go

@@ -67,6 +67,7 @@ func (p *PresenceStreamProvider) IncrementalSync(
 	presences, err := snapshot.PresenceAfter(ctx, from, gomatrixserverlib.EventFilter{Limit: 1000})
 	if err != nil {
 		req.Log.WithError(err).Error("p.DB.PresenceAfter failed")
+		_ = snapshot.Rollback()
 		return from
 	}
 
@@ -95,6 +96,7 @@ func (p *PresenceStreamProvider) IncrementalSync(
 				presences[roomUsers[i]], err = snapshot.GetPresence(ctx, roomUsers[i])
 				if err != nil {
 					req.Log.WithError(err).Error("unable to query presence for user")
+					_ = snapshot.Rollback()
 					return from
 				}
 				if len(presences) > req.Filter.Presence.Limit {

syncapi/streams/stream_receipt.go

@@ -52,6 +52,7 @@ func (p *ReceiptStreamProvider) IncrementalSync(
 	lastPos, receipts, err := snapshot.RoomReceiptsAfter(ctx, joinedRooms, from)
 	if err != nil {
 		req.Log.WithError(err).Error("p.DB.RoomReceiptsAfter failed")
+		_ = snapshot.Rollback()
 		return from
 	}
 

syncapi/streams/stream_sendtodevice.go

@@ -44,6 +44,7 @@ func (p *SendToDeviceStreamProvider) IncrementalSync(
 	lastPos, events, err := snapshot.SendToDeviceUpdatesForSync(req.Context, req.Device.UserID, req.Device.ID, from, to)
 	if err != nil {
 		req.Log.WithError(err).Error("p.DB.SendToDeviceUpdatesForSync failed")
+		_ = snapshot.Rollback()
 		return from
 	}
 

syncapi/streams/streams.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/matrix-org/dendrite/internal/caching"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
 	keyapi "github.com/matrix-org/dendrite/keyserver/api"
 	rsapi "github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/syncapi/notifier"
@@ -72,7 +73,8 @@ func NewSyncStreamProviders(
 	if err != nil {
 		panic(err)
 	}
-	defer snapshot.Rollback() // nolint:errcheck
+	var succeeded bool
+	defer sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err)
 
 	streams.PDUStreamProvider.Setup(ctx, snapshot)
 	streams.TypingStreamProvider.Setup(ctx, snapshot)
@@ -84,6 +86,7 @@ func NewSyncStreamProviders(
 	streams.DeviceListStreamProvider.Setup(ctx, snapshot)
 	streams.PresenceStreamProvider.Setup(ctx, snapshot)
 
+	succeeded = true
 	return streams
 }
 

syncapi/sync/requestpool.go

@@ -31,6 +31,7 @@ import (
 	"github.com/sirupsen/logrus"
 
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
 	keyapi "github.com/matrix-org/dendrite/keyserver/api"
 	roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/setup/config"
@@ -310,7 +311,8 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 			logrus.WithError(err).Error("Failed to acquire database snapshot for sync request")
 			return jsonerror.InternalServerError()
 		}
-		defer snapshot.Rollback() // nolint:errcheck
+		var succeeded bool
+		defer sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err)
 
 		if syncReq.Since.IsEmpty() {
 			// Complete sync
@@ -409,6 +411,7 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 			}
 		}
 
+		succeeded = true
 		return util.JSONResponse{
 			Code: http.StatusOK,
 			JSON: syncReq.Response,
@@ -449,7 +452,8 @@ func (rp *RequestPool) OnIncomingKeyChangeRequest(req *http.Request, device *use
 		logrus.WithError(err).Error("Failed to acquire database snapshot for key change")
 		return jsonerror.InternalServerError()
 	}
-	defer snapshot.Rollback() // nolint:errcheck
+	var succeeded bool
+	defer sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err)
 	rp.streams.PDUStreamProvider.IncrementalSync(req.Context(), snapshot, syncReq, fromToken.PDUPosition, toToken.PDUPosition)
 	_, _, err = internal.DeviceListCatchup(
 		req.Context(), snapshot, rp.keyAPI, rp.rsAPI, syncReq.Device.UserID,
@@ -459,6 +463,7 @@ func (rp *RequestPool) OnIncomingKeyChangeRequest(req *http.Request, device *use
 		util.GetLogger(req.Context()).WithError(err).Error("Failed to DeviceListCatchup info")
 		return jsonerror.InternalServerError()
 	}
+	succeeded = true
 	return util.JSONResponse{
 		Code: 200,
 		JSON: struct {