0
0
Fork 0
mirror of https://github.com/matrix-org/dendrite synced 2024-12-14 21:13:52 +01:00

Reset transaction after a failure

This commit is contained in:
Neil Alexander 2022-09-30 17:07:37 +01:00
parent 16048be236
commit 0116db79c6
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
11 changed files with 40 additions and 17 deletions

View file

@ -29,6 +29,7 @@ import (
type DatabaseTransaction interface { type DatabaseTransaction interface {
sqlutil.Transaction sqlutil.Transaction
Reset() (err error)
SharedUsers SharedUsers
MaxStreamPositionForPDUs(ctx context.Context) (types.StreamPosition, error) MaxStreamPositionForPDUs(ctx context.Context) (types.StreamPosition, error)

View file

@ -77,6 +77,7 @@ func (d *Database) NewDatabaseSnapshot(ctx context.Context) (*DatabaseTransactio
} }
return &DatabaseTransaction{ return &DatabaseTransaction{
Database: d, Database: d,
ctx: ctx,
txn: txn, txn: txn,
}, nil }, nil
*/ */
@ -89,6 +90,7 @@ func (d *Database) NewDatabaseTransaction(ctx context.Context) (*DatabaseTransac
} }
return &DatabaseTransaction{ return &DatabaseTransaction{
Database: d, Database: d,
ctx: ctx,
txn: txn, txn: txn,
}, nil }, nil
} }

View file

@ -13,6 +13,7 @@ import (
type DatabaseTransaction struct { type DatabaseTransaction struct {
*Database *Database
ctx context.Context
txn *sql.Tx txn *sql.Tx
} }
@ -30,6 +31,19 @@ func (d *DatabaseTransaction) Rollback() error {
return d.txn.Rollback() return d.txn.Rollback()
} }
func (d *DatabaseTransaction) Reset() (err error) {
if d.txn == nil {
return nil
}
if err = d.txn.Rollback(); err != nil {
return err
}
if d.txn, err = d.DB.BeginTx(d.ctx, nil); err != nil {
return err
}
return
}
func (d *DatabaseTransaction) MaxStreamPositionForPDUs(ctx context.Context) (types.StreamPosition, error) { func (d *DatabaseTransaction) MaxStreamPositionForPDUs(ctx context.Context) (types.StreamPosition, error) {
id, err := d.OutputEvents.SelectMaxEventID(ctx, d.txn) id, err := d.OutputEvents.SelectMaxEventID(ctx, d.txn)
if err != nil { if err != nil {

View file

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

View file

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

View file

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

View file

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

View file

@ -75,7 +75,7 @@ func (p *PDUStreamProvider) CompleteSync(
joinedRoomIDs, err := snapshot.RoomIDsWithMembership(ctx, req.Device.UserID, gomatrixserverlib.Join) joinedRoomIDs, err := snapshot.RoomIDsWithMembership(ctx, req.Device.UserID, gomatrixserverlib.Join)
if err != nil { if err != nil {
req.Log.WithError(err).Error("p.DB.RoomIDsWithMembership failed") req.Log.WithError(err).Error("p.DB.RoomIDsWithMembership failed")
_ = snapshot.Rollback() _ = snapshot.Reset()
return from return from
} }
@ -102,7 +102,9 @@ func (p *PDUStreamProvider) CompleteSync(
) )
if jerr != nil { if jerr != nil {
req.Log.WithError(jerr).Error("p.getJoinResponseForCompleteSync failed") req.Log.WithError(jerr).Error("p.getJoinResponseForCompleteSync failed")
_ = snapshot.Rollback() if err = snapshot.Reset(); err != nil {
return from
}
continue // return from continue // return from
} }
req.Response.Rooms.Join[roomID] = *jr req.Response.Rooms.Join[roomID] = *jr
@ -113,7 +115,7 @@ func (p *PDUStreamProvider) CompleteSync(
peeks, err := snapshot.PeeksInRange(ctx, req.Device.UserID, req.Device.ID, r) peeks, err := snapshot.PeeksInRange(ctx, req.Device.UserID, req.Device.ID, r)
if err != nil { if err != nil {
req.Log.WithError(err).Error("p.DB.PeeksInRange failed") req.Log.WithError(err).Error("p.DB.PeeksInRange failed")
_ = snapshot.Rollback() _ = snapshot.Reset()
return from return from
} }
for _, peek := range peeks { for _, peek := range peeks {
@ -124,7 +126,9 @@ func (p *PDUStreamProvider) CompleteSync(
) )
if err != nil { if err != nil {
req.Log.WithError(err).Error("p.getJoinResponseForCompleteSync failed") req.Log.WithError(err).Error("p.getJoinResponseForCompleteSync failed")
_ = snapshot.Rollback() if err = snapshot.Reset(); err != nil {
return from
}
continue // return from continue // return from
} }
req.Response.Rooms.Peek[peek.RoomID] = *jr req.Response.Rooms.Peek[peek.RoomID] = *jr
@ -156,13 +160,13 @@ func (p *PDUStreamProvider) IncrementalSync(
if req.WantFullState { if req.WantFullState {
if stateDeltas, syncJoinedRooms, err = snapshot.GetStateDeltasForFullStateSync(ctx, req.Device, r, req.Device.UserID, &stateFilter); err != nil { 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") req.Log.WithError(err).Error("p.DB.GetStateDeltasForFullStateSync failed")
_ = snapshot.Rollback() _ = snapshot.Reset()
return return
} }
} else { } else {
if stateDeltas, syncJoinedRooms, err = snapshot.GetStateDeltas(ctx, req.Device, r, req.Device.UserID, &stateFilter); err != nil { 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") req.Log.WithError(err).Error("p.DB.GetStateDeltas failed")
_ = snapshot.Rollback() _ = snapshot.Reset()
return return
} }
} }
@ -177,7 +181,7 @@ func (p *PDUStreamProvider) IncrementalSync(
if err = p.addIgnoredUsersToFilter(ctx, snapshot, req, &eventFilter); err != nil { if err = p.addIgnoredUsersToFilter(ctx, snapshot, req, &eventFilter); err != nil {
req.Log.WithError(err).Error("unable to update event filter with ignored users") req.Log.WithError(err).Error("unable to update event filter with ignored users")
_ = snapshot.Rollback() _ = snapshot.Reset()
} }
newPos = from newPos = from
@ -197,10 +201,12 @@ func (p *PDUStreamProvider) IncrementalSync(
var pos types.StreamPosition var pos types.StreamPosition
if pos, err = p.addRoomDeltaToResponse(ctx, snapshot, req.Device, newRange, delta, &eventFilter, &stateFilter, req.Response); err != nil { 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") req.Log.WithError(err).Error("d.addRoomDeltaToResponse failed")
_ = snapshot.Rollback()
if err == context.DeadlineExceeded || err == context.Canceled { if err == context.DeadlineExceeded || err == context.Canceled {
return newPos return newPos
} }
if err = snapshot.Reset(); err != nil {
return from
}
continue // return to continue // return to
} }
// Reset the position, as it is only for the special case of newly joined rooms // Reset the position, as it is only for the special case of newly joined rooms
@ -301,7 +307,7 @@ func (p *PDUStreamProvider) addRoomDeltaToResponse(
events, err := applyHistoryVisibilityFilter(ctx, snapshot, p.rsAPI, delta.RoomID, device.UserID, eventFilter.Limit, recentEvents) events, err := applyHistoryVisibilityFilter(ctx, snapshot, p.rsAPI, delta.RoomID, device.UserID, eventFilter.Limit, recentEvents)
if err != nil { if err != nil {
logrus.WithError(err).Error("unable to apply history visibility filter") logrus.WithError(err).Error("unable to apply history visibility filter")
_ = snapshot.Rollback() _ = snapshot.Reset()
} }
if len(delta.StateEvents) > 0 { if len(delta.StateEvents) > 0 {

View file

@ -67,7 +67,7 @@ func (p *PresenceStreamProvider) IncrementalSync(
presences, err := snapshot.PresenceAfter(ctx, from, gomatrixserverlib.EventFilter{Limit: 1000}) presences, err := snapshot.PresenceAfter(ctx, from, gomatrixserverlib.EventFilter{Limit: 1000})
if err != nil { if err != nil {
req.Log.WithError(err).Error("p.DB.PresenceAfter failed") req.Log.WithError(err).Error("p.DB.PresenceAfter failed")
_ = snapshot.Rollback() _ = snapshot.Reset()
return from return from
} }

View file

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

View file

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