aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--syncapi/consumers/roomserver.go5
-rw-r--r--syncapi/notifier/notifier.go9
-rw-r--r--syncapi/routing/context.go5
-rw-r--r--syncapi/routing/search.go5
-rw-r--r--syncapi/storage/interface.go2
-rw-r--r--syncapi/streams/stream_accountdata.go1
-rw-r--r--syncapi/streams/stream_devicelist.go2
-rw-r--r--syncapi/streams/stream_invite.go1
-rw-r--r--syncapi/streams/stream_notificationdata.go1
-rw-r--r--syncapi/streams/stream_pdu.go4
-rw-r--r--syncapi/streams/stream_presence.go2
-rw-r--r--syncapi/streams/stream_receipt.go1
-rw-r--r--syncapi/streams/stream_sendtodevice.go1
-rw-r--r--syncapi/streams/streams.go5
-rw-r--r--syncapi/sync/requestpool.go9
15 files changed, 45 insertions, 8 deletions
diff --git a/syncapi/consumers/roomserver.go b/syncapi/consumers/roomserver.go
index e5e8fe29..c7a11dbb 100644
--- a/syncapi/consumers/roomserver.go
+++ b/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
}
diff --git a/syncapi/notifier/notifier.go b/syncapi/notifier/notifier.go
index a8e5bf9a..db18c6b7 100644
--- a/syncapi/notifier/notifier.go
+++ b/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
}
diff --git a/syncapi/routing/context.go b/syncapi/routing/context.go
index 1ce34b85..0ed164c7 100644
--- a/syncapi/routing/context.go
+++ b/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,
diff --git a/syncapi/routing/search.go b/syncapi/routing/search.go
index bac534a2..aef355de 100644
--- a/syncapi/routing/search.go
+++ b/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,
diff --git a/syncapi/storage/interface.go b/syncapi/storage/interface.go
index 3732e43f..4a03aca7 100644
--- a/syncapi/storage/interface.go
+++ b/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)
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
}
diff --git a/syncapi/sync/requestpool.go b/syncapi/sync/requestpool.go
index 1d0ac1a4..bef0a568 100644
--- a/syncapi/sync/requestpool.go
+++ b/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 {
generated by cgit v1.2.3 (git 2.26.2) at 2024-12-20 07:34:16 +0000