Add constructors for NewXXXUpdater and a useTxns flag

In sqlite, we set useTxns=false and comment why.

roomserver/storage/shared/membership_updater.go

@@ -17,6 +17,46 @@ type membershipUpdater struct {
 	membership    tables.MembershipState
 }
 
+func NewMembershipUpdater(
+	ctx context.Context, d *Database, roomID, targetUserID string,
+	targetLocal bool, roomVersion gomatrixserverlib.RoomVersion,
+	useTxns bool,
+) (types.MembershipUpdater, error) {
+	txn, err := d.DB.Begin()
+	if err != nil {
+		return nil, err
+	}
+	succeeded := false
+	defer func() {
+		if !succeeded {
+			txn.Rollback() // nolint: errcheck
+		}
+	}()
+
+	roomNID, err := d.assignRoomNID(ctx, txn, roomID, roomVersion)
+	if err != nil {
+		return nil, err
+	}
+
+	targetUserNID, err := d.assignStateKeyNID(ctx, txn, targetUserID)
+	if err != nil {
+		return nil, err
+	}
+
+	if !useTxns {
+		txn.Commit() // nolint: errcheck
+		txn = nil
+	}
+
+	updater, err := d.membershipUpdaterTxn(ctx, txn, roomNID, targetUserNID, targetLocal)
+	if err != nil {
+		return nil, err
+	}
+
+	succeeded = true
+	return updater, nil
+}
+
 func (d *Database) membershipUpdaterTxn(
 	ctx context.Context,
 	txn *sql.Tx,

roomserver/storage/shared/room_recent_events_updater.go

@@ -1,6 +1,7 @@
 package shared
 
 import (
+	"context"
 	"database/sql"
 
 	"github.com/matrix-org/dendrite/roomserver/types"
@@ -16,6 +17,39 @@ type roomRecentEventsUpdater struct {
 	currentStateSnapshotNID types.StateSnapshotNID
 }
 
+func NewRoomRecentEventsUpdater(d *Database, ctx context.Context, roomNID types.RoomNID, useTxns bool) (types.RoomRecentEventsUpdater, error) {
+	txn, err := d.DB.Begin()
+	if err != nil {
+		return nil, err
+	}
+	eventNIDs, lastEventNIDSent, currentStateSnapshotNID, err :=
+		d.RoomsTable.SelectLatestEventsNIDsForUpdate(ctx, txn, roomNID)
+	if err != nil {
+		txn.Rollback() // nolint: errcheck
+		return nil, err
+	}
+	stateAndRefs, err := d.EventsTable.BulkSelectStateAtEventAndReference(ctx, txn, eventNIDs)
+	if err != nil {
+		txn.Rollback() // nolint: errcheck
+		return nil, err
+	}
+	var lastEventIDSent string
+	if lastEventNIDSent != 0 {
+		lastEventIDSent, err = d.EventsTable.SelectEventID(ctx, txn, lastEventNIDSent)
+		if err != nil {
+			txn.Rollback() // nolint: errcheck
+			return nil, err
+		}
+	}
+	if !useTxns {
+		txn.Commit() // nolint: errcheck
+		txn = nil
+	}
+	return &roomRecentEventsUpdater{
+		transaction{ctx, txn}, d, roomNID, stateAndRefs, lastEventIDSent, currentStateSnapshotNID,
+	}, nil
+}
+
 // RoomVersion implements types.RoomRecentEventsUpdater
 func (u *roomRecentEventsUpdater) RoomVersion() (version gomatrixserverlib.RoomVersion) {
 	version, _ = u.d.GetRoomVersionForRoomNID(u.ctx, u.roomNID)

roomserver/storage/shared/storage.go

@@ -311,71 +311,17 @@ func (d *Database) GetTransactionEventID(
 	return eventID, err
 }
 
-// MembershipUpdater implements input.RoomEventDatabase
 func (d *Database) MembershipUpdater(
 	ctx context.Context, roomID, targetUserID string,
 	targetLocal bool, roomVersion gomatrixserverlib.RoomVersion,
 ) (types.MembershipUpdater, error) {
-	txn, err := d.DB.Begin()
+	return NewMembershipUpdater(ctx, d, roomID, targetUserID, targetLocal, roomVersion, true)
-	if err != nil {
-		return nil, err
-	}
-	succeeded := false
-	defer func() {
-		if !succeeded {
-			txn.Rollback() // nolint: errcheck
-		}
-	}()
-
-	roomNID, err := d.assignRoomNID(ctx, txn, roomID, roomVersion)
-	if err != nil {
-		return nil, err
-	}
-
-	targetUserNID, err := d.assignStateKeyNID(ctx, txn, targetUserID)
-	if err != nil {
-		return nil, err
-	}
-
-	updater, err := d.membershipUpdaterTxn(ctx, txn, roomNID, targetUserNID, targetLocal)
-	if err != nil {
-		return nil, err
-	}
-
-	succeeded = true
-	return updater, nil
 }
 
-// GetLatestEventsForUpdate implements input.EventDatabase
 func (d *Database) GetLatestEventsForUpdate(
 	ctx context.Context, roomNID types.RoomNID,
 ) (types.RoomRecentEventsUpdater, error) {
-	txn, err := d.DB.Begin()
+	return NewRoomRecentEventsUpdater(d, ctx, roomNID, true)
-	if err != nil {
-		return nil, err
-	}
-	eventNIDs, lastEventNIDSent, currentStateSnapshotNID, err :=
-		d.RoomsTable.SelectLatestEventsNIDsForUpdate(ctx, txn, roomNID)
-	if err != nil {
-		txn.Rollback() // nolint: errcheck
-		return nil, err
-	}
-	stateAndRefs, err := d.EventsTable.BulkSelectStateAtEventAndReference(ctx, txn, eventNIDs)
-	if err != nil {
-		txn.Rollback() // nolint: errcheck
-		return nil, err
-	}
-	var lastEventIDSent string
-	if lastEventNIDSent != 0 {
-		lastEventIDSent, err = d.EventsTable.SelectEventID(ctx, txn, lastEventNIDSent)
-		if err != nil {
-			txn.Rollback() // nolint: errcheck
-			return nil, err
-		}
-	}
-	return &roomRecentEventsUpdater{
-		transaction{ctx, txn}, d, roomNID, stateAndRefs, lastEventIDSent, currentStateSnapshotNID,
-	}, nil
 }
 
 func (d *Database) StoreEvent(

roomserver/storage/sqlite3/storage.go

@@ -140,387 +140,27 @@ func Open(dataSourceName string) (*Database, error) {
 	return &d, nil
 }
 
-func (d *Database) assignRoomNID(
-	ctx context.Context, txn *sql.Tx,
-	roomID string, roomVersion gomatrixserverlib.RoomVersion,
-) (roomNID types.RoomNID, err error) {
-	// Check if we already have a numeric ID in the database.
-	roomNID, err = d.rooms.SelectRoomNID(ctx, txn, roomID)
-	if err == sql.ErrNoRows {
-		// We don't have a numeric ID so insert one into the database.
-		roomNID, err = d.rooms.InsertRoomNID(ctx, txn, roomID, roomVersion)
-		if err == nil {
-			// Now get the numeric ID back out of the database
-			roomNID, err = d.rooms.SelectRoomNID(ctx, txn, roomID)
-		}
-	}
-	return
-}
-
-func (d *Database) assignStateKeyNID(
-	ctx context.Context, txn *sql.Tx, eventStateKey string,
-) (eventStateKeyNID types.EventStateKeyNID, err error) {
-	// Check if we already have a numeric ID in the database.
-	eventStateKeyNID, err = d.eventStateKeys.SelectEventStateKeyNID(ctx, txn, eventStateKey)
-	if err == sql.ErrNoRows {
-		// We don't have a numeric ID so insert one into the database.
-		eventStateKeyNID, err = d.eventStateKeys.InsertEventStateKeyNID(ctx, txn, eventStateKey)
-		if err == sql.ErrNoRows {
-			// We raced with another insert so run the select again.
-			eventStateKeyNID, err = d.eventStateKeys.SelectEventStateKeyNID(ctx, txn, eventStateKey)
-		}
-	}
-	return
-}
-
-// GetLatestEventsForUpdate implements input.EventDatabase
 func (d *Database) GetLatestEventsForUpdate(
 	ctx context.Context, roomNID types.RoomNID,
 ) (types.RoomRecentEventsUpdater, error) {
-	txn, err := d.db.Begin()
+	// TODO: Do not use transactions. We should be holding open this transaction but we cannot have
-	if err != nil {
+	// multiple write transactions on sqlite. The code will perform additional
-		return nil, err
+	// write transactions independent of this one which will consistently cause
-	}
+	// 'database is locked' errors. As sqlite doesn't support multi-process on the
-	eventNIDs, lastEventNIDSent, currentStateSnapshotNID, err :=
+	// same DB anyway, and we only execute updates sequentially, the only worries
-		d.rooms.SelectLatestEventsNIDsForUpdate(ctx, txn, roomNID)
+	// are for rolling back when things go wrong. (atomicity)
-	if err != nil {
+	return shared.NewRoomRecentEventsUpdater(&d.Database, ctx, roomNID, false)
-		txn.Rollback() // nolint: errcheck
-		return nil, err
-	}
-	stateAndRefs, err := d.events.BulkSelectStateAtEventAndReference(ctx, txn, eventNIDs)
-	if err != nil {
-		txn.Rollback() // nolint: errcheck
-		return nil, err
-	}
-	var lastEventIDSent string
-	if lastEventNIDSent != 0 {
-		lastEventIDSent, err = d.events.SelectEventID(ctx, txn, lastEventNIDSent)
-		if err != nil {
-			txn.Rollback() // nolint: errcheck
-			return nil, err
-		}
-	}
-
-	// FIXME: we probably want to support long-lived txns in sqlite somehow, but we don't because we get
-	// 'database is locked' errors caused by multiple write txns (one being the long-lived txn created here)
-	// so for now let's not use a long-lived txn at all, and just commit it here and set the txn to nil so
-	// we fail fast if someone tries to use the underlying txn object.
-	err = txn.Commit()
-	if err != nil {
-		return nil, err
-	}
-	return &roomRecentEventsUpdater{
-		transaction{ctx, nil}, d, roomNID, stateAndRefs, lastEventIDSent, currentStateSnapshotNID,
-	}, nil
 }
 
-type roomRecentEventsUpdater struct {
-	transaction
-	d                       *Database
-	roomNID                 types.RoomNID
-	latestEvents            []types.StateAtEventAndReference
-	lastEventIDSent         string
-	currentStateSnapshotNID types.StateSnapshotNID
-}
-
-// RoomVersion implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) RoomVersion() (version gomatrixserverlib.RoomVersion) {
-	version, _ = u.d.GetRoomVersionForRoomNID(u.ctx, u.roomNID)
-	return
-}
-
-// LatestEvents implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) LatestEvents() []types.StateAtEventAndReference {
-	return u.latestEvents
-}
-
-// LastEventIDSent implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) LastEventIDSent() string {
-	return u.lastEventIDSent
-}
-
-// CurrentStateSnapshotNID implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) CurrentStateSnapshotNID() types.StateSnapshotNID {
-	return u.currentStateSnapshotNID
-}
-
-// StorePreviousEvents implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) StorePreviousEvents(eventNID types.EventNID, previousEventReferences []gomatrixserverlib.EventReference) error {
-	err := internal.WithTransaction(u.d.db, func(txn *sql.Tx) error {
-		for _, ref := range previousEventReferences {
-			if err := u.d.prevEvents.InsertPreviousEvent(u.ctx, txn, ref.EventID, ref.EventSHA256, eventNID); err != nil {
-				return err
-			}
-		}
-		return nil
-	})
-	return err
-}
-
-// IsReferenced implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) IsReferenced(eventReference gomatrixserverlib.EventReference) (res bool, err error) {
-	err = internal.WithTransaction(u.d.db, func(txn *sql.Tx) error {
-		err := u.d.prevEvents.SelectPreviousEventExists(u.ctx, txn, eventReference.EventID, eventReference.EventSHA256)
-		if err == nil {
-			res = true
-			err = nil
-		}
-		if err == sql.ErrNoRows {
-			res = false
-			err = nil
-		}
-		return err
-	})
-	return
-}
-
-// SetLatestEvents implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) SetLatestEvents(
-	roomNID types.RoomNID, latest []types.StateAtEventAndReference, lastEventNIDSent types.EventNID,
-	currentStateSnapshotNID types.StateSnapshotNID,
-) error {
-	err := internal.WithTransaction(u.d.db, func(txn *sql.Tx) error {
-		eventNIDs := make([]types.EventNID, len(latest))
-		for i := range latest {
-			eventNIDs[i] = latest[i].EventNID
-		}
-		return u.d.rooms.UpdateLatestEventNIDs(u.ctx, txn, roomNID, eventNIDs, lastEventNIDSent, currentStateSnapshotNID)
-	})
-	return err
-}
-
-// HasEventBeenSent implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) HasEventBeenSent(eventNID types.EventNID) (res bool, err error) {
-	err = internal.WithTransaction(u.d.db, func(txn *sql.Tx) error {
-		res, err = u.d.events.SelectEventSentToOutput(u.ctx, txn, eventNID)
-		return err
-	})
-	return
-}
-
-// MarkEventAsSent implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) MarkEventAsSent(eventNID types.EventNID) error {
-	err := internal.WithTransaction(u.d.db, func(txn *sql.Tx) error {
-		return u.d.events.UpdateEventSentToOutput(u.ctx, txn, eventNID)
-	})
-	return err
-}
-
-func (u *roomRecentEventsUpdater) MembershipUpdater(targetUserNID types.EventStateKeyNID, targetLocal bool) (mu types.MembershipUpdater, err error) {
-	err = internal.WithTransaction(u.d.db, func(txn *sql.Tx) error {
-		mu, err = u.d.membershipUpdaterTxn(u.ctx, txn, u.roomNID, targetUserNID, targetLocal)
-		return err
-	})
-	return
-}
-
-// MembershipUpdater implements input.RoomEventDatabase
 func (d *Database) MembershipUpdater(
 	ctx context.Context, roomID, targetUserID string,
 	targetLocal bool, roomVersion gomatrixserverlib.RoomVersion,
 ) (updater types.MembershipUpdater, err error) {
-	var txn *sql.Tx
+	// TODO: Do not use transactions. We should be holding open this transaction but we cannot have
-	txn, err = d.db.Begin()
+	// multiple write transactions on sqlite. The code will perform additional
-	if err != nil {
+	// write transactions independent of this one which will consistently cause
-		return nil, err
+	// 'database is locked' errors. As sqlite doesn't support multi-process on the
-	}
+	// same DB anyway, and we only execute updates sequentially, the only worries
-	succeeded := false
+	// are for rolling back when things go wrong. (atomicity)
-	defer func() {
+	return shared.NewMembershipUpdater(ctx, &d.Database, roomID, targetUserID, targetLocal, roomVersion, false)
-		if !succeeded {
-			txn.Rollback() // nolint: errcheck
-		} else {
-			// TODO: We should be holding open this transaction but we cannot have
-			// multiple write transactions on sqlite. The code will perform additional
-			// write transactions independent of this one which will consistently cause
-			// 'database is locked' errors. For now, we'll break up the transaction and
-			// hope we don't race too catastrophically. Long term, we should be able to
-			// thread in txn objects where appropriate (either at the interface level or
-			// bring matrix business logic into the storage layer).
-			txerr := txn.Commit()
-			if err == nil && txerr != nil {
-				err = txerr
-			}
-		}
-	}()
-
-	roomNID, err := d.assignRoomNID(ctx, txn, roomID, roomVersion)
-	if err != nil {
-		return nil, err
-	}
-
-	targetUserNID, err := d.assignStateKeyNID(ctx, txn, targetUserID)
-	if err != nil {
-		return nil, err
-	}
-
-	updater, err = d.membershipUpdaterTxn(ctx, txn, roomNID, targetUserNID, targetLocal)
-	if err != nil {
-		return nil, err
-	}
-
-	succeeded = true
-	return updater, nil
-}
-
-type membershipUpdater struct {
-	transaction
-	d             *Database
-	roomNID       types.RoomNID
-	targetUserNID types.EventStateKeyNID
-	membership    tables.MembershipState
-}
-
-func (d *Database) membershipUpdaterTxn(
-	ctx context.Context,
-	txn *sql.Tx,
-	roomNID types.RoomNID,
-	targetUserNID types.EventStateKeyNID,
-	targetLocal bool,
-) (types.MembershipUpdater, error) {
-
-	if err := d.membership.InsertMembership(ctx, txn, roomNID, targetUserNID, targetLocal); err != nil {
-		return nil, err
-	}
-
-	membership, err := d.membership.SelectMembershipForUpdate(ctx, txn, roomNID, targetUserNID)
-	if err != nil {
-		return nil, err
-	}
-
-	return &membershipUpdater{
-		// purposefully set the txn to nil so if we try to use it we panic and fail fast
-		transaction{ctx, nil}, d, roomNID, targetUserNID, membership,
-	}, nil
-}
-
-// IsInvite implements types.MembershipUpdater
-func (u *membershipUpdater) IsInvite() bool {
-	return u.membership == tables.MembershipStateInvite
-}
-
-// IsJoin implements types.MembershipUpdater
-func (u *membershipUpdater) IsJoin() bool {
-	return u.membership == tables.MembershipStateJoin
-}
-
-// IsLeave implements types.MembershipUpdater
-func (u *membershipUpdater) IsLeave() bool {
-	return u.membership == tables.MembershipStateLeaveOrBan
-}
-
-// SetToInvite implements types.MembershipUpdater
-func (u *membershipUpdater) SetToInvite(event gomatrixserverlib.Event) (inserted bool, err error) {
-	err = internal.WithTransaction(u.d.db, func(txn *sql.Tx) error {
-		senderUserNID, err := u.d.assignStateKeyNID(u.ctx, txn, event.Sender())
-		if err != nil {
-			return err
-		}
-		inserted, err = u.d.invites.InsertInviteEvent(
-			u.ctx, txn, event.EventID(), u.roomNID, u.targetUserNID, senderUserNID, event.JSON(),
-		)
-		if err != nil {
-			return err
-		}
-		if u.membership != tables.MembershipStateInvite {
-			if err = u.d.membership.UpdateMembership(
-				u.ctx, txn, u.roomNID, u.targetUserNID, senderUserNID, tables.MembershipStateInvite, 0,
-			); err != nil {
-				return err
-			}
-		}
-		return nil
-	})
-	return
-}
-
-// SetToJoin implements types.MembershipUpdater
-func (u *membershipUpdater) SetToJoin(senderUserID string, eventID string, isUpdate bool) (inviteEventIDs []string, err error) {
-	err = internal.WithTransaction(u.d.db, func(txn *sql.Tx) error {
-		senderUserNID, err := u.d.assignStateKeyNID(u.ctx, txn, senderUserID)
-		if err != nil {
-			return err
-		}
-
-		// If this is a join event update, there is no invite to update
-		if !isUpdate {
-			inviteEventIDs, err = u.d.invites.UpdateInviteRetired(
-				u.ctx, txn, u.roomNID, u.targetUserNID,
-			)
-			if err != nil {
-				return err
-			}
-		}
-
-		// Look up the NID of the new join event
-		nIDs, err := u.d.EventNIDs(u.ctx, []string{eventID})
-		if err != nil {
-			return err
-		}
-
-		if u.membership != tables.MembershipStateJoin || isUpdate {
-			if err = u.d.membership.UpdateMembership(
-				u.ctx, txn, u.roomNID, u.targetUserNID, senderUserNID,
-				tables.MembershipStateJoin, nIDs[eventID],
-			); err != nil {
-				return err
-			}
-		}
-		return nil
-	})
-
-	return
-}
-
-// SetToLeave implements types.MembershipUpdater
-func (u *membershipUpdater) SetToLeave(senderUserID string, eventID string) (inviteEventIDs []string, err error) {
-	err = internal.WithTransaction(u.d.db, func(txn *sql.Tx) error {
-		senderUserNID, err := u.d.assignStateKeyNID(u.ctx, txn, senderUserID)
-		if err != nil {
-			return err
-		}
-		inviteEventIDs, err = u.d.invites.UpdateInviteRetired(
-			u.ctx, txn, u.roomNID, u.targetUserNID,
-		)
-		if err != nil {
-			return err
-		}
-
-		// Look up the NID of the new leave event
-		nIDs, err := u.d.EventNIDs(u.ctx, []string{eventID})
-		if err != nil {
-			return err
-		}
-
-		if u.membership != tables.MembershipStateLeaveOrBan {
-			if err = u.d.membership.UpdateMembership(
-				u.ctx, txn, u.roomNID, u.targetUserNID, senderUserNID,
-				tables.MembershipStateLeaveOrBan, nIDs[eventID],
-			); err != nil {
-				return err
-			}
-		}
-		return nil
-	})
-	return
-}
-
-type transaction struct {
-	ctx context.Context
-	txn *sql.Tx
-}
-
-// Commit implements types.Transaction
-func (t *transaction) Commit() error {
-	if t.txn == nil {
-		return nil
-	}
-	return t.txn.Commit()
-}
-
-// Rollback implements types.Transaction
-func (t *transaction) Rollback() error {
-	if t.txn == nil {
-		return nil
-	}
-	return t.txn.Rollback()
 }