Fix broken `/sync` due to transaction error

9 files changed, 17 insertions, 1 deletions

diff --git a/syncapi/streams/stream_accountdata.go b/syncapi/streams/stream_accountdata.go
index 3f2f7d13..34135d65 100644
--- a/syncapi/streams/stream_accountdata.go
+++ b/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
 	}
 
diff --git a/syncapi/streams/stream_devicelist.go b/syncapi/streams/stream_devicelist.go
index 7996c203..10ede573 100644
--- a/syncapi/streams/stream_devicelist.go
+++ b/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
 	}
 
diff --git a/syncapi/streams/stream_invite.go b/syncapi/streams/stream_invite.go
index 17b3b843..b52eaaab 100644
--- a/syncapi/streams/stream_invite.go
+++ b/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
 	}
 
diff --git a/syncapi/streams/stream_notificationdata.go b/syncapi/streams/stream_notificationdata.go
index 5a81fd09..e1ee02b2 100644
--- a/syncapi/streams/stream_notificationdata.go
+++ b/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
 	}
 
diff --git a/syncapi/streams/stream_pdu.go b/syncapi/streams/stream_pdu.go
index 89c5ba35..43e2ce8b 100644
--- a/syncapi/streams/stream_pdu.go
+++ b/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
diff --git a/syncapi/streams/stream_presence.go b/syncapi/streams/stream_presence.go
index 81cea7d5..d24c8562 100644
--- a/syncapi/streams/stream_presence.go
+++ b/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 {
diff --git a/syncapi/streams/stream_receipt.go b/syncapi/streams/stream_receipt.go
index 8818a553..40e5bd01 100644
--- a/syncapi/streams/stream_receipt.go
+++ b/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
 	}
 
diff --git a/syncapi/streams/stream_sendtodevice.go b/syncapi/streams/stream_sendtodevice.go
index 00b67cc4..3262832a 100644
--- a/syncapi/streams/stream_sendtodevice.go
+++ b/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
 	}
 
diff --git a/syncapi/streams/streams.go b/syncapi/streams/streams.go
index eccbb3a4..dc854762 100644
--- a/syncapi/streams/streams.go
+++ b/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
 }