Modify sync transaction behaviour (#2758)

This now uses a transaction per stream, so that errors in one stream
don't propagate to another, and we therefore no longer need to do hacks
to reopen a new transaction after aborting a failed one.
This commit is contained in:
Neil Alexander 2022-10-03 11:38:20 +01:00 committed by GitHub
parent d4710217f8
commit d32f60249d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 155 additions and 91 deletions

View File

@ -29,7 +29,6 @@ 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

@ -31,19 +31,6 @@ 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,6 @@ 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.Reset()
return from return from
} }

View File

@ -34,13 +34,11 @@ 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.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.Reset()
return from return from
} }

View File

@ -56,7 +56,6 @@ 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.Reset()
return from return from
} }

View File

@ -46,7 +46,6 @@ 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.Reset()
return from return from
} }

View File

@ -75,7 +75,6 @@ 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.Reset()
return from return from
} }
@ -102,10 +101,10 @@ 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")
if err = snapshot.Reset(); err != nil { if err == context.DeadlineExceeded || err == context.Canceled || err == sql.ErrTxDone {
return from return from
} }
continue // return from continue
} }
req.Response.Rooms.Join[roomID] = *jr req.Response.Rooms.Join[roomID] = *jr
req.Rooms[roomID] = gomatrixserverlib.Join req.Rooms[roomID] = gomatrixserverlib.Join
@ -115,7 +114,6 @@ 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.Reset()
return from return from
} }
for _, peek := range peeks { for _, peek := range peeks {
@ -126,10 +124,10 @@ 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")
if err = snapshot.Reset(); err != nil { if err == context.DeadlineExceeded || err == context.Canceled || err == sql.ErrTxDone {
return from return from
} }
continue // return from continue
} }
req.Response.Rooms.Peek[peek.RoomID] = *jr req.Response.Rooms.Peek[peek.RoomID] = *jr
} }
@ -160,14 +158,12 @@ 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.Reset() return from
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.Reset() return from
return
} }
} }
@ -181,7 +177,6 @@ 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.Reset()
} }
newPos = from newPos = from
@ -201,13 +196,10 @@ 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")
if err == context.DeadlineExceeded || err == context.Canceled { if err == context.DeadlineExceeded || err == context.Canceled || err == sql.ErrTxDone {
return newPos return newPos
} }
if err = snapshot.Reset(); err != nil { continue
return from
}
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
if delta.NewlyJoined { if delta.NewlyJoined {
@ -307,7 +299,6 @@ 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.Reset()
} }
if len(delta.StateEvents) > 0 { if len(delta.StateEvents) > 0 {

View File

@ -67,7 +67,6 @@ 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.Reset()
return from return from
} }

View File

@ -52,7 +52,6 @@ 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.Reset()
return from return from
} }

View File

@ -44,7 +44,6 @@ 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.Reset()
return from return from
} }

View File

@ -306,13 +306,19 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
syncReq.Log.WithField("currentPos", currentPos).Debugln("Responding to sync immediately") syncReq.Log.WithField("currentPos", currentPos).Debugln("Responding to sync immediately")
} }
snapshot, err := rp.db.NewDatabaseSnapshot(req.Context()) withTransaction := func(from types.StreamPosition, f func(snapshot storage.DatabaseTransaction) types.StreamPosition) types.StreamPosition {
if err != nil { var succeeded bool
logrus.WithError(err).Error("Failed to acquire database snapshot for sync request") snapshot, err := rp.db.NewDatabaseSnapshot(req.Context())
return jsonerror.InternalServerError() if err != nil {
logrus.WithError(err).Error("Failed to acquire database snapshot for sync request")
return from
}
defer func() {
succeeded = err == nil
sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err)
}()
return f(snapshot)
} }
var succeeded bool
defer sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err)
if syncReq.Since.IsEmpty() { if syncReq.Since.IsEmpty() {
// Complete sync // Complete sync
@ -320,72 +326,162 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
// Get the current DeviceListPosition first, as the currentPosition // Get the current DeviceListPosition first, as the currentPosition
// might advance while processing other streams, resulting in flakey // might advance while processing other streams, resulting in flakey
// tests. // tests.
DeviceListPosition: rp.streams.DeviceListStreamProvider.CompleteSync( DeviceListPosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.DeviceListPosition,
func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.DeviceListStreamProvider.CompleteSync(
syncReq.Context, txn, syncReq,
)
},
), ),
PDUPosition: rp.streams.PDUStreamProvider.CompleteSync( PDUPosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.PDUPosition,
func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.PDUStreamProvider.CompleteSync(
syncReq.Context, txn, syncReq,
)
},
), ),
TypingPosition: rp.streams.TypingStreamProvider.CompleteSync( TypingPosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.TypingPosition,
func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.TypingStreamProvider.CompleteSync(
syncReq.Context, txn, syncReq,
)
},
), ),
ReceiptPosition: rp.streams.ReceiptStreamProvider.CompleteSync( ReceiptPosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.ReceiptPosition,
func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.ReceiptStreamProvider.CompleteSync(
syncReq.Context, txn, syncReq,
)
},
), ),
InvitePosition: rp.streams.InviteStreamProvider.CompleteSync( InvitePosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.InvitePosition,
func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.InviteStreamProvider.CompleteSync(
syncReq.Context, txn, syncReq,
)
},
), ),
SendToDevicePosition: rp.streams.SendToDeviceStreamProvider.CompleteSync( SendToDevicePosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.SendToDevicePosition,
func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.SendToDeviceStreamProvider.CompleteSync(
syncReq.Context, txn, syncReq,
)
},
), ),
AccountDataPosition: rp.streams.AccountDataStreamProvider.CompleteSync( AccountDataPosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.AccountDataPosition,
func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.AccountDataStreamProvider.CompleteSync(
syncReq.Context, txn, syncReq,
)
},
), ),
NotificationDataPosition: rp.streams.NotificationDataStreamProvider.CompleteSync( NotificationDataPosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.NotificationDataPosition,
func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.NotificationDataStreamProvider.CompleteSync(
syncReq.Context, txn, syncReq,
)
},
), ),
PresencePosition: rp.streams.PresenceStreamProvider.CompleteSync( PresencePosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.PresencePosition,
func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.PresenceStreamProvider.CompleteSync(
syncReq.Context, txn, syncReq,
)
},
), ),
} }
} else { } else {
// Incremental sync // Incremental sync
syncReq.Response.NextBatch = types.StreamingToken{ syncReq.Response.NextBatch = types.StreamingToken{
PDUPosition: rp.streams.PDUStreamProvider.IncrementalSync( PDUPosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.PDUPosition,
syncReq.Since.PDUPosition, currentPos.PDUPosition, func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.PDUStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq,
syncReq.Since.PDUPosition, currentPos.PDUPosition,
)
},
), ),
TypingPosition: rp.streams.TypingStreamProvider.IncrementalSync( TypingPosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.TypingPosition,
syncReq.Since.TypingPosition, currentPos.TypingPosition, func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.TypingStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq,
syncReq.Since.TypingPosition, currentPos.TypingPosition,
)
},
), ),
ReceiptPosition: rp.streams.ReceiptStreamProvider.IncrementalSync( ReceiptPosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.ReceiptPosition,
syncReq.Since.ReceiptPosition, currentPos.ReceiptPosition, func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.ReceiptStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq,
syncReq.Since.ReceiptPosition, currentPos.ReceiptPosition,
)
},
), ),
InvitePosition: rp.streams.InviteStreamProvider.IncrementalSync( InvitePosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.InvitePosition,
syncReq.Since.InvitePosition, currentPos.InvitePosition, func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.InviteStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq,
syncReq.Since.InvitePosition, currentPos.InvitePosition,
)
},
), ),
SendToDevicePosition: rp.streams.SendToDeviceStreamProvider.IncrementalSync( SendToDevicePosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.SendToDevicePosition,
syncReq.Since.SendToDevicePosition, currentPos.SendToDevicePosition, func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.SendToDeviceStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq,
syncReq.Since.SendToDevicePosition, currentPos.SendToDevicePosition,
)
},
), ),
AccountDataPosition: rp.streams.AccountDataStreamProvider.IncrementalSync( AccountDataPosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.AccountDataPosition,
syncReq.Since.AccountDataPosition, currentPos.AccountDataPosition, func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.AccountDataStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq,
syncReq.Since.AccountDataPosition, currentPos.AccountDataPosition,
)
},
), ),
NotificationDataPosition: rp.streams.NotificationDataStreamProvider.IncrementalSync( NotificationDataPosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.NotificationDataPosition,
syncReq.Since.NotificationDataPosition, currentPos.NotificationDataPosition, func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.NotificationDataStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq,
syncReq.Since.NotificationDataPosition, currentPos.NotificationDataPosition,
)
},
), ),
DeviceListPosition: rp.streams.DeviceListStreamProvider.IncrementalSync( DeviceListPosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.DeviceListPosition,
syncReq.Since.DeviceListPosition, currentPos.DeviceListPosition, func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.DeviceListStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq,
syncReq.Since.DeviceListPosition, currentPos.DeviceListPosition,
)
},
), ),
PresencePosition: rp.streams.PresenceStreamProvider.IncrementalSync( PresencePosition: withTransaction(
syncReq.Context, snapshot, syncReq, syncReq.Since.PresencePosition,
syncReq.Since.PresencePosition, currentPos.PresencePosition, func(txn storage.DatabaseTransaction) types.StreamPosition {
return rp.streams.PresenceStreamProvider.IncrementalSync(
syncReq.Context, txn, syncReq,
syncReq.Since.PresencePosition, currentPos.PresencePosition,
)
},
), ),
} }
// it's possible for there to be no updates for this user even though since < current pos, // it's possible for there to be no updates for this user even though since < current pos,
@ -411,7 +507,6 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
} }
} }
succeeded = true
return util.JSONResponse{ return util.JSONResponse{
Code: http.StatusOK, Code: http.StatusOK,
JSON: syncReq.Response, JSON: syncReq.Response,