Merge branch 'master' into kegan/fedsender-as-fedclient-subset

appservice/storage/sqlite3/appservice_events_table.go

@@ -67,7 +67,7 @@ const (
 
 type eventsStatements struct {
 	db                                     *sql.DB
-	writer                                 *sqlutil.TransactionWriter
+	writer                                 sqlutil.TransactionWriter
 	selectEventsByApplicationServiceIDStmt *sql.Stmt
 	countEventsByApplicationServiceIDStmt  *sql.Stmt
 	insertEventStmt                        *sql.Stmt

appservice/storage/sqlite3/txn_id_counter_table.go

@@ -38,7 +38,7 @@ const selectTxnIDSQL = `
 
 type txnStatements struct {
 	db              *sql.DB
-	writer          *sqlutil.TransactionWriter
+	writer          sqlutil.TransactionWriter
 	selectTxnIDStmt *sql.Stmt
 }
 

clientapi/routing/sendevent.go

@@ -16,6 +16,7 @@ package routing
 
 import (
 	"net/http"
+	"sync"
 
 	"github.com/matrix-org/dendrite/clientapi/httputil"
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
@@ -35,6 +36,10 @@ type sendEventResponse struct {
 	EventID string `json:"event_id"`
 }
 
+var (
+	userRoomSendMutexes sync.Map // (roomID+userID) -> mutex. mutexes to ensure correct ordering of sendEvents
+)
+
 // SendEvent implements:
 //   /rooms/{roomID}/send/{eventType}
 //   /rooms/{roomID}/send/{eventType}/{txnID}
@@ -63,6 +68,13 @@ func SendEvent(
 		}
 	}
 
+	// create a mutex for the specific user in the specific room
+	// this avoids a situation where events that are received in quick succession are sent to the roomserver in a jumbled order
+	userID := device.UserID
+	mutex, _ := userRoomSendMutexes.LoadOrStore(roomID+userID, &sync.Mutex{})
+	mutex.(*sync.Mutex).Lock()
+	defer mutex.(*sync.Mutex).Unlock()
+
 	e, resErr := generateSendEvent(req, device, roomID, eventType, stateKey, cfg, rsAPI)
 	if resErr != nil {
 		return *resErr

currentstateserver/storage/sqlite3/current_room_state_table.go

@@ -83,7 +83,7 @@ const selectKnownUsersSQL = "" +
 
 type currentRoomStateStatements struct {
 	db                               *sql.DB
-	writer                           *sqlutil.TransactionWriter
+	writer                           sqlutil.TransactionWriter
 	upsertRoomStateStmt              *sql.Stmt
 	deleteRoomStateByEventIDStmt     *sql.Stmt
 	selectRoomIDsWithMembershipStmt  *sql.Stmt

federationsender/statistics/statistics.go

@@ -66,10 +66,16 @@ type ServerStatistics struct {
 	serverName     gomatrixserverlib.ServerName //
 	blacklisted    atomic.Bool                  // is the node blacklisted
 	backoffStarted atomic.Bool                  // is the backoff started
+	backoffUntil   atomic.Value                 // time.Time until this backoff interval ends
 	backoffCount   atomic.Uint32                // number of times BackoffDuration has been called
 	successCounter atomic.Uint32                // how many times have we succeeded?
 }
 
+// duration returns how long the next backoff interval should be.
+func (s *ServerStatistics) duration(count uint32) time.Duration {
+	return time.Second * time.Duration(math.Exp2(float64(count)))
+}
+
 // Success updates the server statistics with a new successful
 // attempt, which increases the sent counter and resets the idle and
 // failure counters. If a host was blacklisted at this point then
@@ -88,11 +94,36 @@ func (s *ServerStatistics) Success() {
 
 // Failure marks a failure and starts backing off if needed.
 // The next call to BackoffIfRequired will do the right thing
-// after this.
+// after this. It will return the time that the current failure
-func (s *ServerStatistics) Failure() {
+// will result in backoff waiting until, and a bool signalling
+// whether we have blacklisted and therefore to give up.
+func (s *ServerStatistics) Failure() (time.Time, bool) {
+	// If we aren't already backing off, this call will start
+	// a new backoff period. Reset the counter to 0 so that
+	// we backoff only for short periods of time to start with.
 	if s.backoffStarted.CAS(false, true) {
 		s.backoffCount.Store(0)
 	}
+
+	// Check if we have blacklisted this node.
+	if s.blacklisted.Load() {
+		return time.Now(), true
+	}
+
+	// If we're already backing off and we haven't yet surpassed
+	// the deadline then return that. Repeated calls to Failure
+	// within a single backoff interval will have no side effects.
+	if until, ok := s.backoffUntil.Load().(time.Time); ok && !time.Now().After(until) {
+		return until, false
+	}
+
+	// We're either backing off and have passed the deadline, or
+	// we aren't backing off, so work out what the next interval
+	// will be.
+	count := s.backoffCount.Load()
+	until := time.Now().Add(s.duration(count))
+	s.backoffUntil.Store(until)
+	return until, false
 }
 
 // BackoffIfRequired will block for as long as the current
@@ -103,21 +134,8 @@ func (s *ServerStatistics) BackoffIfRequired(backingOff atomic.Bool, interrupt <
 		return 0, false
 	}
 
-	// Check if the interrupt channel is already closed. If it is
-	// then this call should have no side effects but still return
-	// the current duration.
-	select {
-	case <-interrupt:
-		count := s.backoffCount.Load()
-		return time.Second * time.Duration(math.Exp2(float64(count))), false
-	default:
-	}
-
-	// Work out how many times we've backed off so far.
-	count := s.backoffCount.Inc()
-	duration := time.Second * time.Duration(math.Exp2(float64(count)))
-
 	// Work out if we should be blacklisting at this point.
+	count := s.backoffCount.Inc()
 	if count >= s.statistics.FailuresUntilBlacklist {
 		// We've exceeded the maximum amount of times we're willing
 		// to back off, which is probably in the region of hours by
@@ -129,9 +147,14 @@ func (s *ServerStatistics) BackoffIfRequired(backingOff atomic.Bool, interrupt <
 				logrus.WithError(err).Errorf("Failed to add %q to blacklist", s.serverName)
 			}
 		}
-		return duration, true
+		return 0, true
 	}
 
+	// Work out when we should wait until.
+	duration := s.duration(count)
+	until := time.Now().Add(duration)
+	s.backoffUntil.Store(until)
+
 	// Notify the destination queue that we're backing off now.
 	backingOff.Store(true)
 	defer backingOff.Store(false)

federationsender/statistics/statistics_test.go

@@ -10,7 +10,7 @@ import (
 
 func TestBackoff(t *testing.T) {
 	stats := Statistics{
-		FailuresUntilBlacklist: 5,
+		FailuresUntilBlacklist: 7,
 	}
 	server := ServerStatistics{
 		statistics: &stats,
@@ -44,10 +44,20 @@ func TestBackoff(t *testing.T) {
 		// Get the duration.
 		duration, blacklist := server.BackoffIfRequired(backingOff, interrupt)
 
+		// Register another failure for good measure. This should have no
+		// side effects since a backoff is already in progress. If it does
+		// then we'll fail.
+		until, blacklisted := server.Failure()
+		if time.Until(until) > duration {
+			t.Fatal("Failure produced unexpected side effect when it shouldn't have")
+		}
+
 		// Check if we should be blacklisted by now.
-		if i > stats.FailuresUntilBlacklist {
+		if i >= stats.FailuresUntilBlacklist {
 			if !blacklist {
 				t.Fatalf("Backoff %d should have resulted in blacklist but didn't", i)
+			} else if blacklist != blacklisted {
+				t.Fatalf("BackoffIfRequired and Failure returned different blacklist values")
 			} else {
 				t.Logf("Backoff %d is blacklisted as expected", i)
 				continue

federationsender/storage/postgres/blacklist_table.go

@@ -42,7 +42,6 @@ const deleteBlacklistSQL = "" +
 
 type blacklistStatements struct {
 	db                  *sql.DB
-	writer              *sqlutil.TransactionWriter
 	insertBlacklistStmt *sql.Stmt
 	selectBlacklistStmt *sql.Stmt
 	deleteBlacklistStmt *sql.Stmt
@@ -50,8 +49,7 @@ type blacklistStatements struct {
 
 func NewPostgresBlacklistTable(db *sql.DB) (s *blacklistStatements, err error) {
 	s = &blacklistStatements{
-		db:     db,
+		db: db,
-		writer: sqlutil.NewTransactionWriter(),
 	}
 	_, err = db.Exec(blacklistSchema)
 	if err != nil {
@@ -75,11 +73,9 @@ func NewPostgresBlacklistTable(db *sql.DB) (s *blacklistStatements, err error) {
 func (s *blacklistStatements) InsertBlacklist(
 	ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName,
 ) error {
-	return s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	stmt := sqlutil.TxStmt(txn, s.insertBlacklistStmt)
-		stmt := sqlutil.TxStmt(txn, s.insertBlacklistStmt)
+	_, err := stmt.ExecContext(ctx, serverName)
-		_, err := stmt.ExecContext(ctx, serverName)
+	return err
-		return err
-	})
 }
 
 // selectRoomForUpdate locks the row for the room and returns the last_event_id.
@@ -105,9 +101,7 @@ func (s *blacklistStatements) SelectBlacklist(
 func (s *blacklistStatements) DeleteBlacklist(
 	ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName,
 ) error {
-	return s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	stmt := sqlutil.TxStmt(txn, s.deleteBlacklistStmt)
-		stmt := sqlutil.TxStmt(txn, s.deleteBlacklistStmt)
+	_, err := stmt.ExecContext(ctx, serverName)
-		_, err := stmt.ExecContext(ctx, serverName)
+	return err
-		return err
-	})
 }

federationsender/storage/sqlite3/blacklist_table.go

@@ -42,7 +42,7 @@ const deleteBlacklistSQL = "" +
 
 type blacklistStatements struct {
 	db                  *sql.DB
-	writer              *sqlutil.TransactionWriter
+	writer              sqlutil.TransactionWriter
 	insertBlacklistStmt *sql.Stmt
 	selectBlacklistStmt *sql.Stmt
 	deleteBlacklistStmt *sql.Stmt

federationsender/storage/sqlite3/joined_hosts_table.go

@@ -65,7 +65,7 @@ const selectJoinedHostsForRoomsSQL = "" +
 
 type joinedHostsStatements struct {
 	db                       *sql.DB
-	writer                   *sqlutil.TransactionWriter
+	writer                   sqlutil.TransactionWriter
 	insertJoinedHostsStmt    *sql.Stmt
 	deleteJoinedHostsStmt    *sql.Stmt
 	selectJoinedHostsStmt    *sql.Stmt

federationsender/storage/sqlite3/queue_edus_table.go

@@ -64,7 +64,7 @@ const selectQueueServerNamesSQL = "" +
 
 type queueEDUsStatements struct {
 	db                                   *sql.DB
-	writer                               *sqlutil.TransactionWriter
+	writer                               sqlutil.TransactionWriter
 	insertQueueEDUStmt                   *sql.Stmt
 	selectQueueEDUStmt                   *sql.Stmt
 	selectQueueEDUReferenceJSONCountStmt *sql.Stmt

federationsender/storage/sqlite3/queue_json_table.go

@@ -50,7 +50,7 @@ const selectJSONSQL = "" +
 
 type queueJSONStatements struct {
 	db             *sql.DB
-	writer         *sqlutil.TransactionWriter
+	writer         sqlutil.TransactionWriter
 	insertJSONStmt *sql.Stmt
 	//deleteJSONStmt *sql.Stmt - prepared at runtime due to variadic
 	//selectJSONStmt *sql.Stmt - prepared at runtime due to variadic

federationsender/storage/sqlite3/queue_pdus_table.go

@@ -71,7 +71,7 @@ const selectQueuePDUsServerNamesSQL = "" +
 
 type queuePDUsStatements struct {
 	db                                *sql.DB
-	writer                            *sqlutil.TransactionWriter
+	writer                            sqlutil.TransactionWriter
 	insertQueuePDUStmt                *sql.Stmt
 	selectQueueNextTransactionIDStmt  *sql.Stmt
 	selectQueuePDUsByTransactionStmt  *sql.Stmt

federationsender/storage/sqlite3/room_table.go

@@ -44,7 +44,7 @@ const updateRoomSQL = "" +
 
 type roomStatements struct {
 	db                      *sql.DB
-	writer                  *sqlutil.TransactionWriter
+	writer                  sqlutil.TransactionWriter
 	insertRoomStmt          *sql.Stmt
 	selectRoomForUpdateStmt *sql.Stmt
 	updateRoomStmt          *sql.Stmt

internal/sqlutil/sql.go

@@ -19,8 +19,6 @@ import (
 	"errors"
 	"fmt"
 	"runtime"
-
-	"go.uber.org/atomic"
 )
 
 // ErrUserExists is returned if a username already exists in the database.
@@ -52,7 +50,7 @@ func EndTransaction(txn Transaction, succeeded *bool) error {
 func WithTransaction(db *sql.DB, fn func(txn *sql.Tx) error) (err error) {
 	txn, err := db.Begin()
 	if err != nil {
-		return
+		return fmt.Errorf("sqlutil.WithTransaction.Begin: %w", err)
 	}
 	succeeded := false
 	defer func() {
@@ -106,69 +104,6 @@ func SQLiteDriverName() string {
 	return "sqlite3"
 }
 
-// TransactionWriter allows queuing database writes so that you don't
+type TransactionWriter interface {
-// contend on database locks in, e.g. SQLite. Only one task will run
+	Do(db *sql.DB, txn *sql.Tx, f func(txn *sql.Tx) error) error
-// at a time on a given TransactionWriter.
-type TransactionWriter struct {
-	running atomic.Bool
-	todo    chan transactionWriterTask
-}
-
-func NewTransactionWriter() *TransactionWriter {
-	return &TransactionWriter{
-		todo: make(chan transactionWriterTask),
-	}
-}
-
-// transactionWriterTask represents a specific task.
-type transactionWriterTask struct {
-	db   *sql.DB
-	txn  *sql.Tx
-	f    func(txn *sql.Tx) error
-	wait chan error
-}
-
-// Do queues a task to be run by a TransactionWriter. The function
-// provided will be ran within a transaction as supplied by the
-// txn parameter if one is supplied, and if not, will take out a
-// new transaction from the database supplied in the database
-// parameter. Either way, this will block until the task is done.
-func (w *TransactionWriter) Do(db *sql.DB, txn *sql.Tx, f func(txn *sql.Tx) error) error {
-	if w.todo == nil {
-		return errors.New("not initialised")
-	}
-	if !w.running.Load() {
-		go w.run()
-	}
-	task := transactionWriterTask{
-		db:   db,
-		txn:  txn,
-		f:    f,
-		wait: make(chan error, 1),
-	}
-	w.todo <- task
-	return <-task.wait
-}
-
-// run processes the tasks for a given transaction writer. Only one
-// of these goroutines will run at a time. A transaction will be
-// opened using the database object from the task and then this will
-// be passed as a parameter to the task function.
-func (w *TransactionWriter) run() {
-	if !w.running.CAS(false, true) {
-		return
-	}
-	defer w.running.Store(false)
-	for task := range w.todo {
-		if task.txn != nil {
-			task.wait <- task.f(task.txn)
-		} else if task.db != nil {
-			task.wait <- WithTransaction(task.db, func(txn *sql.Tx) error {
-				return task.f(txn)
-			})
-		} else {
-			panic("expected database or transaction but got neither")
-		}
-		close(task.wait)
-	}
 }

internal/sqlutil/writer_dummy.go

@@ -0,0 +1,22 @@
+package sqlutil
+
+import (
+	"database/sql"
+)
+
+type DummyTransactionWriter struct {
+}
+
+func NewDummyTransactionWriter() TransactionWriter {
+	return &DummyTransactionWriter{}
+}
+
+func (w *DummyTransactionWriter) Do(db *sql.DB, txn *sql.Tx, f func(txn *sql.Tx) error) error {
+	if txn == nil {
+		return WithTransaction(db, func(txn *sql.Tx) error {
+			return f(txn)
+		})
+	} else {
+		return f(txn)
+	}
+}

internal/sqlutil/writer_exclusive.go

@@ -0,0 +1,75 @@
+package sqlutil
+
+import (
+	"database/sql"
+	"errors"
+
+	"go.uber.org/atomic"
+)
+
+// ExclusiveTransactionWriter allows queuing database writes so that you don't
+// contend on database locks in, e.g. SQLite. Only one task will run
+// at a time on a given ExclusiveTransactionWriter.
+type ExclusiveTransactionWriter struct {
+	running atomic.Bool
+	todo    chan transactionWriterTask
+}
+
+func NewTransactionWriter() TransactionWriter {
+	return &ExclusiveTransactionWriter{
+		todo: make(chan transactionWriterTask),
+	}
+}
+
+// transactionWriterTask represents a specific task.
+type transactionWriterTask struct {
+	db   *sql.DB
+	txn  *sql.Tx
+	f    func(txn *sql.Tx) error
+	wait chan error
+}
+
+// Do queues a task to be run by a TransactionWriter. The function
+// provided will be ran within a transaction as supplied by the
+// txn parameter if one is supplied, and if not, will take out a
+// new transaction from the database supplied in the database
+// parameter. Either way, this will block until the task is done.
+func (w *ExclusiveTransactionWriter) Do(db *sql.DB, txn *sql.Tx, f func(txn *sql.Tx) error) error {
+	if w.todo == nil {
+		return errors.New("not initialised")
+	}
+	if !w.running.Load() {
+		go w.run()
+	}
+	task := transactionWriterTask{
+		db:   db,
+		txn:  txn,
+		f:    f,
+		wait: make(chan error, 1),
+	}
+	w.todo <- task
+	return <-task.wait
+}
+
+// run processes the tasks for a given transaction writer. Only one
+// of these goroutines will run at a time. A transaction will be
+// opened using the database object from the task and then this will
+// be passed as a parameter to the task function.
+func (w *ExclusiveTransactionWriter) run() {
+	if !w.running.CAS(false, true) {
+		return
+	}
+	defer w.running.Store(false)
+	for task := range w.todo {
+		if task.txn != nil {
+			task.wait <- task.f(task.txn)
+		} else if task.db != nil {
+			task.wait <- WithTransaction(task.db, func(txn *sql.Tx) error {
+				return task.f(txn)
+			})
+		} else {
+			panic("expected database or transaction but got neither")
+		}
+		close(task.wait)
+	}
+}

keyserver/storage/sqlite3/device_keys_table.go

@@ -63,7 +63,7 @@ const deleteAllDeviceKeysSQL = "" +
 
 type deviceKeysStatements struct {
 	db                         *sql.DB
-	writer                     *sqlutil.TransactionWriter
+	writer                     sqlutil.TransactionWriter
 	upsertDeviceKeysStmt       *sql.Stmt
 	selectDeviceKeysStmt       *sql.Stmt
 	selectBatchDeviceKeysStmt  *sql.Stmt

keyserver/storage/sqlite3/key_changes_table.go

@@ -52,7 +52,7 @@ const selectKeyChangesSQL = "" +
 
 type keyChangesStatements struct {
 	db                   *sql.DB
-	writer               *sqlutil.TransactionWriter
+	writer               sqlutil.TransactionWriter
 	upsertKeyChangeStmt  *sql.Stmt
 	selectKeyChangesStmt *sql.Stmt
 }

keyserver/storage/sqlite3/one_time_keys_table.go

@@ -60,7 +60,7 @@ const selectKeyByAlgorithmSQL = "" +
 
 type oneTimeKeysStatements struct {
 	db                       *sql.DB
-	writer                   *sqlutil.TransactionWriter
+	writer                   sqlutil.TransactionWriter
 	upsertKeysStmt           *sql.Stmt
 	selectKeysStmt           *sql.Stmt
 	selectKeysCountStmt      *sql.Stmt

mediaapi/storage/sqlite3/media_repository_table.go

@@ -62,7 +62,7 @@ SELECT content_type, file_size_bytes, creation_ts, upload_name, base64hash, user
 
 type mediaStatements struct {
 	db              *sql.DB
-	writer          *sqlutil.TransactionWriter
+	writer          sqlutil.TransactionWriter
 	insertMediaStmt *sql.Stmt
 	selectMediaStmt *sql.Stmt
 }

roomserver/internal/input_latest_events.go

@@ -24,6 +24,7 @@ import (
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/roomserver/state"
+	"github.com/matrix-org/dendrite/roomserver/storage/shared"
 	"github.com/matrix-org/dendrite/roomserver/types"
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/matrix-org/util"
@@ -56,7 +57,7 @@ func (r *RoomserverInternalAPI) updateLatestEvents(
 ) (err error) {
 	updater, err := r.DB.GetLatestEventsForUpdate(ctx, roomNID)
 	if err != nil {
-		return
+		return fmt.Errorf("r.DB.GetLatestEventsForUpdate: %w", err)
 	}
 	succeeded := false
 	defer func() {
@@ -78,7 +79,7 @@ func (r *RoomserverInternalAPI) updateLatestEvents(
 	}
 
 	if err = u.doUpdateLatestEvents(); err != nil {
-		return err
+		return fmt.Errorf("u.doUpdateLatestEvents: %w", err)
 	}
 
 	succeeded = true
@@ -92,7 +93,7 @@ func (r *RoomserverInternalAPI) updateLatestEvents(
 type latestEventsUpdater struct {
 	ctx           context.Context
 	api           *RoomserverInternalAPI
-	updater       types.RoomRecentEventsUpdater
+	updater       *shared.LatestEventsUpdater
 	roomNID       types.RoomNID
 	stateAtEvent  types.StateAtEvent
 	event         gomatrixserverlib.Event
@@ -136,7 +137,7 @@ func (u *latestEventsUpdater) doUpdateLatestEvents() error {
 	// don't need to do anything, as we've handled it already.
 	hasBeenSent, err := u.updater.HasEventBeenSent(u.stateAtEvent.EventNID)
 	if err != nil {
-		return err
+		return fmt.Errorf("u.updater.HasEventBeenSent: %w", err)
 	} else if hasBeenSent {
 		return nil
 	}
@@ -144,7 +145,7 @@ func (u *latestEventsUpdater) doUpdateLatestEvents() error {
 	// Update the roomserver_previous_events table with references. This
 	// is effectively tracking the structure of the DAG.
 	if err = u.updater.StorePreviousEvents(u.stateAtEvent.EventNID, prevEvents); err != nil {
-		return err
+		return fmt.Errorf("u.updater.StorePreviousEvents: %w", err)
 	}
 
 	// Get the event reference for our new event. This will be used when
@@ -155,7 +156,7 @@ func (u *latestEventsUpdater) doUpdateLatestEvents() error {
 	// in the room. If it is then it isn't a latest event.
 	alreadyReferenced, err := u.updater.IsReferenced(eventReference)
 	if err != nil {
-		return err
+		return fmt.Errorf("u.updater.IsReferenced: %w", err)
 	}
 
 	// Work out what the latest events are.
@@ -172,19 +173,19 @@ func (u *latestEventsUpdater) doUpdateLatestEvents() error {
 	// Now that we know what the latest events are, it's time to get the
 	// latest state.
 	if err = u.latestState(); err != nil {
-		return err
+		return fmt.Errorf("u.latestState: %w", err)
 	}
 
 	// If we need to generate any output events then here's where we do it.
 	// TODO: Move this!
 	updates, err := u.api.updateMemberships(u.ctx, u.updater, u.removed, u.added)
 	if err != nil {
-		return err
+		return fmt.Errorf("u.api.updateMemberships: %w", err)
 	}
 
 	update, err := u.makeOutputNewRoomEvent()
 	if err != nil {
-		return err
+		return fmt.Errorf("u.makeOutputNewRoomEvent: %w", err)
 	}
 	updates = append(updates, *update)
 
@@ -197,14 +198,18 @@ func (u *latestEventsUpdater) doUpdateLatestEvents() error {
 	// the correct order, 2) that pending writes are resent across restarts. In order to avoid writing all the
 	// necessary bookkeeping we'll keep the event sending synchronous for now.
 	if err = u.api.WriteOutputEvents(u.event.RoomID(), updates); err != nil {
-		return err
+		return fmt.Errorf("u.api.WriteOutputEvents: %w", err)
 	}
 
 	if err = u.updater.SetLatestEvents(u.roomNID, u.latest, u.stateAtEvent.EventNID, u.newStateNID); err != nil {
-		return err
+		return fmt.Errorf("u.updater.SetLatestEvents: %w", err)
 	}
 
-	return u.updater.MarkEventAsSent(u.stateAtEvent.EventNID)
+	if err = u.updater.MarkEventAsSent(u.stateAtEvent.EventNID); err != nil {
+		return fmt.Errorf("u.updater.MarkEventAsSent: %w", err)
+	}
+
+	return nil
 }
 
 func (u *latestEventsUpdater) latestState() error {
@@ -224,7 +229,7 @@ func (u *latestEventsUpdater) latestState() error {
 		u.ctx, u.roomNID, latestStateAtEvents,
 	)
 	if err != nil {
-		return err
+		return fmt.Errorf("roomState.CalculateAndStoreStateAfterEvents: %w", err)
 	}
 
 	// If we are overwriting the state then we should make sure that we
@@ -243,7 +248,7 @@ func (u *latestEventsUpdater) latestState() error {
 		u.ctx, u.oldStateNID, u.newStateNID,
 	)
 	if err != nil {
-		return err
+		return fmt.Errorf("roomState.DifferenceBetweenStateSnapshots: %w", err)
 	}
 
 	// Also work out the state before the event removes and the event
@@ -251,7 +256,11 @@ func (u *latestEventsUpdater) latestState() error {
 	u.stateBeforeEventRemoves, u.stateBeforeEventAdds, err = roomState.DifferenceBetweeenStateSnapshots(
 		u.ctx, u.newStateNID, u.stateAtEvent.BeforeStateSnapshotNID,
 	)
-	return err
+	if err != nil {
+		return fmt.Errorf("roomState.DifferenceBetweeenStateSnapshots: %w", err)
+	}
+
+	return nil
 }
 
 func calculateLatest(

roomserver/internal/input_membership.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 
 	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/roomserver/storage/shared"
 	"github.com/matrix-org/dendrite/roomserver/types"
 	"github.com/matrix-org/gomatrixserverlib"
 )
@@ -29,7 +30,7 @@ import (
 // consumers about the invites added or retired by the change in current state.
 func (r *RoomserverInternalAPI) updateMemberships(
 	ctx context.Context,
-	updater types.RoomRecentEventsUpdater,
+	updater *shared.LatestEventsUpdater,
 	removed, added []types.StateEntry,
 ) ([]api.OutputEvent, error) {
 	changes := membershipChanges(removed, added)
@@ -77,7 +78,7 @@ func (r *RoomserverInternalAPI) updateMemberships(
 }
 
 func (r *RoomserverInternalAPI) updateMembership(
-	updater types.RoomRecentEventsUpdater,
+	updater *shared.LatestEventsUpdater,
 	targetUserNID types.EventStateKeyNID,
 	remove, add *gomatrixserverlib.Event,
 	updates []api.OutputEvent,
@@ -141,7 +142,7 @@ func (r *RoomserverInternalAPI) isLocalTarget(event *gomatrixserverlib.Event) bo
 }
 
 func updateToInviteMembership(
-	mu types.MembershipUpdater, add *gomatrixserverlib.Event, updates []api.OutputEvent,
+	mu *shared.MembershipUpdater, add *gomatrixserverlib.Event, updates []api.OutputEvent,
 	roomVersion gomatrixserverlib.RoomVersion,
 ) ([]api.OutputEvent, error) {
 	// We may have already sent the invite to the user, either because we are
@@ -171,7 +172,7 @@ func updateToInviteMembership(
 }
 
 func updateToJoinMembership(
-	mu types.MembershipUpdater, add *gomatrixserverlib.Event, updates []api.OutputEvent,
+	mu *shared.MembershipUpdater, add *gomatrixserverlib.Event, updates []api.OutputEvent,
 ) ([]api.OutputEvent, error) {
 	// If the user is already marked as being joined, we call SetToJoin to update
 	// the event ID then we can return immediately. Retired is ignored as there
@@ -207,7 +208,7 @@ func updateToJoinMembership(
 }
 
 func updateToLeaveMembership(
-	mu types.MembershipUpdater, add *gomatrixserverlib.Event,
+	mu *shared.MembershipUpdater, add *gomatrixserverlib.Event,
 	newMembership string, updates []api.OutputEvent,
 ) ([]api.OutputEvent, error) {
 	// If the user is already neither joined, nor invited to the room then we

roomserver/state/state.go

@@ -558,7 +558,11 @@ func (v StateResolution) CalculateAndStoreStateAfterEvents(
 		// 2) There weren't any prev_events for this event so the state is
 		// empty.
 		metrics.algorithm = "empty_state"
-		return metrics.stop(v.db.AddState(ctx, roomNID, nil, nil))
+		stateNID, err := v.db.AddState(ctx, roomNID, nil, nil)
+		if err != nil {
+			err = fmt.Errorf("v.db.AddState: %w", err)
+		}
+		return metrics.stop(stateNID, err)
 	}
 
 	if len(prevStates) == 1 {
@@ -578,22 +582,30 @@ func (v StateResolution) CalculateAndStoreStateAfterEvents(
 		)
 		if err != nil {
 			metrics.algorithm = "_load_state_blocks"
-			return metrics.stop(0, err)
+			return metrics.stop(0, fmt.Errorf("v.db.StateBlockNIDs: %w", err))
 		}
 		stateBlockNIDs := stateBlockNIDLists[0].StateBlockNIDs
 		if len(stateBlockNIDs) < maxStateBlockNIDs {
 			// 4) The number of state data blocks is small enough that we can just
 			// add the state event as a block of size one to the end of the blocks.
 			metrics.algorithm = "single_delta"
-			return metrics.stop(v.db.AddState(
+			stateNID, err := v.db.AddState(
 				ctx, roomNID, stateBlockNIDs, []types.StateEntry{prevState.StateEntry},
-			))
+			)
+			if err != nil {
+				err = fmt.Errorf("v.db.AddState: %w", err)
+			}
+			return metrics.stop(stateNID, err)
 		}
 		// If there are too many deltas then we need to calculate the full state
 		// So fall through to calculateAndStoreStateAfterManyEvents
 	}
 
-	return v.calculateAndStoreStateAfterManyEvents(ctx, roomNID, prevStates, metrics)
+	stateNID, err := v.calculateAndStoreStateAfterManyEvents(ctx, roomNID, prevStates, metrics)
+	if err != nil {
+		return 0, fmt.Errorf("v.calculateAndStoreStateAfterManyEvents: %w", err)
+	}
+	return stateNID, nil
 }
 
 // maxStateBlockNIDs is the maximum number of state data blocks to use to encode a snapshot of room state.

roomserver/storage/interface.go

@@ -18,6 +18,7 @@ import (
 	"context"
 
 	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/roomserver/storage/shared"
 	"github.com/matrix-org/dendrite/roomserver/types"
 	"github.com/matrix-org/gomatrixserverlib"
 )
@@ -86,7 +87,7 @@ type Database interface {
 	// The RoomRecentEventsUpdater must have Commit or Rollback called on it if this doesn't return an error.
 	// Returns the latest events in the room and the last eventID sent to the log along with an updater.
 	// If this returns an error then no further action is required.
-	GetLatestEventsForUpdate(ctx context.Context, roomNID types.RoomNID) (types.RoomRecentEventsUpdater, error)
+	GetLatestEventsForUpdate(ctx context.Context, roomNID types.RoomNID) (*shared.LatestEventsUpdater, error)
 	// Look up event ID by transaction's info.
 	// This is used to determine if the room event is processed/processing already.
 	// Returns an empty string if no such event exists.
@@ -123,7 +124,7 @@ type Database interface {
 	// Returns an error if there was a problem talking to the database.
 	RemoveRoomAlias(ctx context.Context, alias string) error
 	// Build a membership updater for the target user in a room.
-	MembershipUpdater(ctx context.Context, roomID, targetUserID string, targetLocal bool, roomVersion gomatrixserverlib.RoomVersion) (types.MembershipUpdater, error)
+	MembershipUpdater(ctx context.Context, roomID, targetUserID string, targetLocal bool, roomVersion gomatrixserverlib.RoomVersion) (*shared.MembershipUpdater, error)
 	// Lookup the membership of a given user in a given room.
 	// Returns the numeric ID of the latest membership event sent from this user
 	// in this room, along a boolean set to true if the user is still in this room,

roomserver/storage/postgres/storage.go

@@ -98,6 +98,7 @@ func Open(dbProperties *config.DatabaseOptions) (*Database, error) {
 	}
 	d.Database = shared.Database{
 		DB:                  db,
+		Writer:              sqlutil.NewDummyTransactionWriter(),
 		EventTypesTable:     eventTypes,
 		EventStateKeysTable: eventStateKeys,
 		EventJSONTable:      eventJSON,

roomserver/storage/shared/latest_events_updater.go

@@ -3,12 +3,13 @@ package shared
 import (
 	"context"
 	"database/sql"
+	"fmt"
 
 	"github.com/matrix-org/dendrite/roomserver/types"
 	"github.com/matrix-org/gomatrixserverlib"
 )
 
-type roomRecentEventsUpdater struct {
+type LatestEventsUpdater struct {
 	transaction
 	d                       *Database
 	roomNID                 types.RoomNID
@@ -17,11 +18,7 @@ type roomRecentEventsUpdater struct {
 	currentStateSnapshotNID types.StateSnapshotNID
 }
 
-func NewRoomRecentEventsUpdater(d *Database, ctx context.Context, roomNID types.RoomNID, useTxns bool) (types.RoomRecentEventsUpdater, error) {
+func NewLatestEventsUpdater(ctx context.Context, d *Database, txn *sql.Tx, roomNID types.RoomNID) (*LatestEventsUpdater, 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 {
@@ -41,48 +38,46 @@ func NewRoomRecentEventsUpdater(d *Database, ctx context.Context, roomNID types.
 			return nil, err
 		}
 	}
-	if !useTxns {
+	return &LatestEventsUpdater{
-		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) {
+func (u *LatestEventsUpdater) RoomVersion() (version gomatrixserverlib.RoomVersion) {
 	version, _ = u.d.GetRoomVersionForRoomNID(u.ctx, u.roomNID)
 	return
 }
 
 // LatestEvents implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) LatestEvents() []types.StateAtEventAndReference {
+func (u *LatestEventsUpdater) LatestEvents() []types.StateAtEventAndReference {
 	return u.latestEvents
 }
 
 // LastEventIDSent implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) LastEventIDSent() string {
+func (u *LatestEventsUpdater) LastEventIDSent() string {
 	return u.lastEventIDSent
 }
 
 // CurrentStateSnapshotNID implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) CurrentStateSnapshotNID() types.StateSnapshotNID {
+func (u *LatestEventsUpdater) CurrentStateSnapshotNID() types.StateSnapshotNID {
 	return u.currentStateSnapshotNID
 }
 
 // StorePreviousEvents implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) StorePreviousEvents(eventNID types.EventNID, previousEventReferences []gomatrixserverlib.EventReference) error {
+func (u *LatestEventsUpdater) StorePreviousEvents(eventNID types.EventNID, previousEventReferences []gomatrixserverlib.EventReference) error {
-	for _, ref := range previousEventReferences {
+	return u.d.Writer.Do(u.d.DB, u.txn, func(txn *sql.Tx) error {
-		if err := u.d.PrevEventsTable.InsertPreviousEvent(u.ctx, u.txn, ref.EventID, ref.EventSHA256, eventNID); err != nil {
+		for _, ref := range previousEventReferences {
-			return err
+			if err := u.d.PrevEventsTable.InsertPreviousEvent(u.ctx, txn, ref.EventID, ref.EventSHA256, eventNID); err != nil {
+				return fmt.Errorf("u.d.PrevEventsTable.InsertPreviousEvent: %w", err)
+			}
 		}
-	}
+		return nil
-	return nil
+	})
 }
 
 // IsReferenced implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) IsReferenced(eventReference gomatrixserverlib.EventReference) (bool, error) {
+func (u *LatestEventsUpdater) IsReferenced(eventReference gomatrixserverlib.EventReference) (bool, error) {
 	err := u.d.PrevEventsTable.SelectPreviousEventExists(u.ctx, u.txn, eventReference.EventID, eventReference.EventSHA256)
 	if err == nil {
 		return true, nil
@@ -90,11 +85,11 @@ func (u *roomRecentEventsUpdater) IsReferenced(eventReference gomatrixserverlib.
 	if err == sql.ErrNoRows {
 		return false, nil
 	}
-	return false, err
+	return false, fmt.Errorf("u.d.PrevEventsTable.SelectPreviousEventExists: %w", err)
 }
 
 // SetLatestEvents implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) SetLatestEvents(
+func (u *LatestEventsUpdater) SetLatestEvents(
 	roomNID types.RoomNID, latest []types.StateAtEventAndReference, lastEventNIDSent types.EventNID,
 	currentStateSnapshotNID types.StateSnapshotNID,
 ) error {
@@ -102,19 +97,26 @@ func (u *roomRecentEventsUpdater) SetLatestEvents(
 	for i := range latest {
 		eventNIDs[i] = latest[i].EventNID
 	}
-	return u.d.RoomsTable.UpdateLatestEventNIDs(u.ctx, u.txn, roomNID, eventNIDs, lastEventNIDSent, currentStateSnapshotNID)
+	return u.d.Writer.Do(u.d.DB, u.txn, func(txn *sql.Tx) error {
+		if err := u.d.RoomsTable.UpdateLatestEventNIDs(u.ctx, txn, roomNID, eventNIDs, lastEventNIDSent, currentStateSnapshotNID); err != nil {
+			return fmt.Errorf("u.d.RoomsTable.updateLatestEventNIDs: %w", err)
+		}
+		return nil
+	})
 }
 
 // HasEventBeenSent implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) HasEventBeenSent(eventNID types.EventNID) (bool, error) {
+func (u *LatestEventsUpdater) HasEventBeenSent(eventNID types.EventNID) (bool, error) {
 	return u.d.EventsTable.SelectEventSentToOutput(u.ctx, u.txn, eventNID)
 }
 
 // MarkEventAsSent implements types.RoomRecentEventsUpdater
-func (u *roomRecentEventsUpdater) MarkEventAsSent(eventNID types.EventNID) error {
+func (u *LatestEventsUpdater) MarkEventAsSent(eventNID types.EventNID) error {
-	return u.d.EventsTable.UpdateEventSentToOutput(u.ctx, u.txn, eventNID)
+	return u.d.Writer.Do(u.d.DB, u.txn, func(txn *sql.Tx) error {
+		return u.d.EventsTable.UpdateEventSentToOutput(u.ctx, txn, eventNID)
+	})
 }
 
-func (u *roomRecentEventsUpdater) MembershipUpdater(targetUserNID types.EventStateKeyNID, targetLocal bool) (types.MembershipUpdater, error) {
+func (u *LatestEventsUpdater) MembershipUpdater(targetUserNID types.EventStateKeyNID, targetLocal bool) (*MembershipUpdater, error) {
 	return u.d.membershipUpdaterTxn(u.ctx, u.txn, u.roomNID, targetUserNID, targetLocal)
 }

roomserver/storage/shared/membership_updater.go

@@ -3,13 +3,14 @@ package shared
 import (
 	"context"
 	"database/sql"
+	"fmt"
 
 	"github.com/matrix-org/dendrite/roomserver/storage/tables"
 	"github.com/matrix-org/dendrite/roomserver/types"
 	"github.com/matrix-org/gomatrixserverlib"
 )
 
-type membershipUpdater struct {
+type MembershipUpdater struct {
 	transaction
 	d             *Database
 	roomNID       types.RoomNID
@@ -18,21 +19,9 @@ type membershipUpdater struct {
 }
 
 func NewMembershipUpdater(
-	ctx context.Context, d *Database, roomID, targetUserID string,
+	ctx context.Context, d *Database, txn *sql.Tx, roomID, targetUserID string,
 	targetLocal bool, roomVersion gomatrixserverlib.RoomVersion,
-	useTxns bool,
+) (*MembershipUpdater, error) {
-) (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
@@ -43,17 +32,7 @@ func NewMembershipUpdater(
 		return nil, err
 	}
 
-	updater, err := d.membershipUpdaterTxn(ctx, txn, roomNID, targetUserNID, targetLocal)
+	return d.membershipUpdaterTxn(ctx, txn, roomNID, targetUserNID, targetLocal)
-	if err != nil {
-		return nil, err
-	}
-
-	succeeded = true
-	if !useTxns {
-		txn.Commit() // nolint: errcheck
-		updater.transaction.txn = nil
-	}
-	return updater, nil
 }
 
 func (d *Database) membershipUpdaterTxn(
@@ -62,10 +41,15 @@ func (d *Database) membershipUpdaterTxn(
 	roomNID types.RoomNID,
 	targetUserNID types.EventStateKeyNID,
 	targetLocal bool,
-) (*membershipUpdater, error) {
+) (*MembershipUpdater, error) {
-
+	err := d.Writer.Do(d.DB, txn, func(txn *sql.Tx) error {
-	if err := d.MembershipTable.InsertMembership(ctx, txn, roomNID, targetUserNID, targetLocal); err != nil {
+		if err := d.MembershipTable.InsertMembership(ctx, txn, roomNID, targetUserNID, targetLocal); err != nil {
-		return nil, err
+			return fmt.Errorf("d.MembershipTable.InsertMembership: %w", err)
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("u.d.Writer.Do: %w", err)
 	}
 
 	membership, err := d.MembershipTable.SelectMembershipForUpdate(ctx, txn, roomNID, targetUserNID)
@@ -73,55 +57,55 @@ func (d *Database) membershipUpdaterTxn(
 		return nil, err
 	}
 
-	return &membershipUpdater{
+	return &MembershipUpdater{
 		transaction{ctx, txn}, d, roomNID, targetUserNID, membership,
 	}, nil
 }
 
 // IsInvite implements types.MembershipUpdater
-func (u *membershipUpdater) IsInvite() bool {
+func (u *MembershipUpdater) IsInvite() bool {
 	return u.membership == tables.MembershipStateInvite
 }
 
 // IsJoin implements types.MembershipUpdater
-func (u *membershipUpdater) IsJoin() bool {
+func (u *MembershipUpdater) IsJoin() bool {
 	return u.membership == tables.MembershipStateJoin
 }
 
 // IsLeave implements types.MembershipUpdater
-func (u *membershipUpdater) IsLeave() bool {
+func (u *MembershipUpdater) IsLeave() bool {
 	return u.membership == tables.MembershipStateLeaveOrBan
 }
 
 // SetToInvite implements types.MembershipUpdater
-func (u *membershipUpdater) SetToInvite(event gomatrixserverlib.Event) (bool, error) {
+func (u *MembershipUpdater) SetToInvite(event gomatrixserverlib.Event) (bool, error) {
 	senderUserNID, err := u.d.assignStateKeyNID(u.ctx, u.txn, event.Sender())
 	if err != nil {
-		return false, err
+		return false, fmt.Errorf("u.d.AssignStateKeyNID: %w", err)
 	}
 	inserted, err := u.d.InvitesTable.InsertInviteEvent(
 		u.ctx, u.txn, event.EventID(), u.roomNID, u.targetUserNID, senderUserNID, event.JSON(),
 	)
 	if err != nil {
-		return false, err
+		return false, fmt.Errorf("u.d.InvitesTable.InsertInviteEvent: %w", err)
 	}
 	if u.membership != tables.MembershipStateInvite {
 		if err = u.d.MembershipTable.UpdateMembership(
 			u.ctx, u.txn, u.roomNID, u.targetUserNID, senderUserNID, tables.MembershipStateInvite, 0,
 		); err != nil {
-			return false, err
+			return false, fmt.Errorf("u.d.MembershipTable.UpdateMembership: %w", err)
 		}
 	}
 	return inserted, nil
 }
 
 // SetToJoin implements types.MembershipUpdater
-func (u *membershipUpdater) SetToJoin(senderUserID string, eventID string, isUpdate bool) ([]string, error) {
+func (u *MembershipUpdater) SetToJoin(senderUserID string, eventID string, isUpdate bool) ([]string, error) {
 	var inviteEventIDs []string
 
 	senderUserNID, err := u.d.assignStateKeyNID(u.ctx, u.txn, senderUserID)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("u.d.AssignStateKeyNID: %w", err)
 	}
 
 	// If this is a join event update, there is no invite to update
@@ -130,14 +114,14 @@ func (u *membershipUpdater) SetToJoin(senderUserID string, eventID string, isUpd
 			u.ctx, u.txn, u.roomNID, u.targetUserNID,
 		)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("u.d.InvitesTables.UpdateInviteRetired: %w", err)
 		}
 	}
 
 	// Look up the NID of the new join event
 	nIDs, err := u.d.EventNIDs(u.ctx, []string{eventID})
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("u.d.EventNIDs: %w", err)
 	}
 
 	if u.membership != tables.MembershipStateJoin || isUpdate {
@@ -145,7 +129,7 @@ func (u *membershipUpdater) SetToJoin(senderUserID string, eventID string, isUpd
 			u.ctx, u.txn, u.roomNID, u.targetUserNID, senderUserNID,
 			tables.MembershipStateJoin, nIDs[eventID],
 		); err != nil {
-			return nil, err
+			return nil, fmt.Errorf("u.d.MembershipTable.UpdateMembership: %w", err)
 		}
 	}
 
@@ -153,22 +137,22 @@ func (u *membershipUpdater) SetToJoin(senderUserID string, eventID string, isUpd
 }
 
 // SetToLeave implements types.MembershipUpdater
-func (u *membershipUpdater) SetToLeave(senderUserID string, eventID string) ([]string, error) {
+func (u *MembershipUpdater) SetToLeave(senderUserID string, eventID string) ([]string, error) {
 	senderUserNID, err := u.d.assignStateKeyNID(u.ctx, u.txn, senderUserID)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("u.d.AssignStateKeyNID: %w", err)
 	}
 	inviteEventIDs, err := u.d.InvitesTable.UpdateInviteRetired(
 		u.ctx, u.txn, u.roomNID, u.targetUserNID,
 	)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("u.d.InvitesTable.updateInviteRetired: %w", err)
 	}
 
 	// Look up the NID of the new leave event
 	nIDs, err := u.d.EventNIDs(u.ctx, []string{eventID})
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("u.d.EventNIDs: %w", err)
 	}
 
 	if u.membership != tables.MembershipStateLeaveOrBan {
@@ -176,7 +160,7 @@ func (u *membershipUpdater) SetToLeave(senderUserID string, eventID string) ([]s
 			u.ctx, u.txn, u.roomNID, u.targetUserNID, senderUserNID,
 			tables.MembershipStateLeaveOrBan, nIDs[eventID],
 		); err != nil {
-			return nil, err
+			return nil, fmt.Errorf("u.d.MembershipTable.UpdateMembership: %w", err)
 		}
 	}
 	return inviteEventIDs, nil

roomserver/storage/shared/storage.go

@@ -27,6 +27,7 @@ const redactionsArePermanent = false
 
 type Database struct {
 	DB                  *sql.DB
+	Writer              sqlutil.TransactionWriter
 	EventsTable         tables.Events
 	EventJSONTable      tables.EventJSON
 	EventTypesTable     tables.EventTypes
@@ -83,20 +84,23 @@ func (d *Database) AddState(
 	stateBlockNIDs []types.StateBlockNID,
 	state []types.StateEntry,
 ) (stateNID types.StateSnapshotNID, err error) {
-	err = sqlutil.WithTransaction(d.DB, func(txn *sql.Tx) error {
+	err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
 		if len(state) > 0 {
 			var stateBlockNID types.StateBlockNID
 			stateBlockNID, err = d.StateBlockTable.BulkInsertStateData(ctx, txn, state)
 			if err != nil {
-				return err
+				return fmt.Errorf("d.StateBlockTable.BulkInsertStateData: %w", err)
 			}
 			stateBlockNIDs = append(stateBlockNIDs[:len(stateBlockNIDs):len(stateBlockNIDs)], stateBlockNID)
 		}
 		stateNID, err = d.StateSnapshotTable.InsertState(ctx, txn, roomNID, stateBlockNIDs)
-		return err
+		if err != nil {
+			return fmt.Errorf("d.StateSnapshotTable.InsertState: %w", err)
+		}
+		return nil
 	})
 	if err != nil {
-		return 0, err
+		return 0, fmt.Errorf("d.Writer.Do: %w", err)
 	}
 	return
 }
@@ -110,7 +114,9 @@ func (d *Database) EventNIDs(
 func (d *Database) SetState(
 	ctx context.Context, eventNID types.EventNID, stateNID types.StateSnapshotNID,
 ) error {
-	return d.EventsTable.UpdateEventState(ctx, eventNID, stateNID)
+	return d.Writer.Do(d.DB, nil, func(_ *sql.Tx) error {
+		return d.EventsTable.UpdateEventState(ctx, eventNID, stateNID)
+	})
 }
 
 func (d *Database) StateAtEventIDs(
@@ -173,22 +179,19 @@ func (d *Database) RoomNIDExcludingStubs(ctx context.Context, roomID string) (ro
 func (d *Database) LatestEventIDs(
 	ctx context.Context, roomNID types.RoomNID,
 ) (references []gomatrixserverlib.EventReference, currentStateSnapshotNID types.StateSnapshotNID, depth int64, err error) {
-	err = sqlutil.WithTransaction(d.DB, func(txn *sql.Tx) error {
+	var eventNIDs []types.EventNID
-		var eventNIDs []types.EventNID
+	eventNIDs, currentStateSnapshotNID, err = d.RoomsTable.SelectLatestEventNIDs(ctx, nil, roomNID)
-		eventNIDs, currentStateSnapshotNID, err = d.RoomsTable.SelectLatestEventNIDs(ctx, txn, roomNID)
+	if err != nil {
-		if err != nil {
+		return
-			return err
+	}
-		}
+	references, err = d.EventsTable.BulkSelectEventReference(ctx, nil, eventNIDs)
-		references, err = d.EventsTable.BulkSelectEventReference(ctx, txn, eventNIDs)
+	if err != nil {
-		if err != nil {
+		return
-			return err
+	}
-		}
+	depth, err = d.EventsTable.SelectMaxEventDepth(ctx, nil, eventNIDs)
-		depth, err = d.EventsTable.SelectMaxEventDepth(ctx, txn, eventNIDs)
+	if err != nil {
-		if err != nil {
+		return
-			return err
+	}
-		}
-		return nil
-	})
 	return
 }
 
@@ -221,7 +224,9 @@ func (d *Database) GetRoomVersionForRoomNID(
 }
 
 func (d *Database) SetRoomAlias(ctx context.Context, alias string, roomID string, creatorUserID string) error {
-	return d.RoomAliasesTable.InsertRoomAlias(ctx, alias, roomID, creatorUserID)
+	return d.Writer.Do(d.DB, nil, func(_ *sql.Tx) error {
+		return d.RoomAliasesTable.InsertRoomAlias(ctx, alias, roomID, creatorUserID)
+	})
 }
 
 func (d *Database) GetRoomIDForAlias(ctx context.Context, alias string) (string, error) {
@@ -239,15 +244,21 @@ func (d *Database) GetCreatorIDForAlias(
 }
 
 func (d *Database) RemoveRoomAlias(ctx context.Context, alias string) error {
-	return d.RoomAliasesTable.DeleteRoomAlias(ctx, alias)
+	return d.Writer.Do(d.DB, nil, func(_ *sql.Tx) error {
+		return d.RoomAliasesTable.DeleteRoomAlias(ctx, alias)
+	})
 }
 
 func (d *Database) GetMembership(
 	ctx context.Context, roomNID types.RoomNID, requestSenderUserID string,
 ) (membershipEventNID types.EventNID, stillInRoom bool, err error) {
-	requestSenderUserNID, err := d.assignStateKeyNID(ctx, nil, requestSenderUserID)
+	var requestSenderUserNID types.EventStateKeyNID
+	err = d.Writer.Do(d.DB, nil, func(_ *sql.Tx) error {
+		requestSenderUserNID, err = d.assignStateKeyNID(ctx, nil, requestSenderUserID)
+		return err
+	})
 	if err != nil {
-		return
+		return 0, false, fmt.Errorf("d.assignStateKeyNID: %w", err)
 	}
 
 	senderMembershipEventNID, senderMembership, err :=
@@ -332,16 +343,35 @@ func (d *Database) GetTransactionEventID(
 func (d *Database) MembershipUpdater(
 	ctx context.Context, roomID, targetUserID string,
 	targetLocal bool, roomVersion gomatrixserverlib.RoomVersion,
-) (types.MembershipUpdater, error) {
+) (*MembershipUpdater, error) {
-	return NewMembershipUpdater(ctx, d, roomID, targetUserID, targetLocal, roomVersion, true)
+	txn, err := d.DB.Begin()
+	if err != nil {
+		return nil, err
+	}
+	var updater *MembershipUpdater
+	_ = d.Writer.Do(d.DB, txn, func(txn *sql.Tx) error {
+		updater, err = NewMembershipUpdater(ctx, d, txn, roomID, targetUserID, targetLocal, roomVersion)
+		return nil
+	})
+	return updater, err
 }
 
 func (d *Database) GetLatestEventsForUpdate(
 	ctx context.Context, roomNID types.RoomNID,
-) (types.RoomRecentEventsUpdater, error) {
+) (*LatestEventsUpdater, error) {
-	return NewRoomRecentEventsUpdater(d, ctx, roomNID, true)
+	txn, err := d.DB.Begin()
+	if err != nil {
+		return nil, err
+	}
+	var updater *LatestEventsUpdater
+	_ = d.Writer.Do(d.DB, txn, func(txn *sql.Tx) error {
+		updater, err = NewLatestEventsUpdater(ctx, d, txn, roomNID)
+		return nil
+	})
+	return updater, err
 }
 
+// nolint:gocyclo
 func (d *Database) StoreEvent(
 	ctx context.Context, event gomatrixserverlib.Event,
 	txnAndSessionID *api.TransactionID, authEventNIDs []types.EventNID,
@@ -357,7 +387,7 @@ func (d *Database) StoreEvent(
 		err              error
 	)
 
-	err = sqlutil.WithTransaction(d.DB, func(txn *sql.Tx) error {
+	err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
 		if txnAndSessionID != nil {
 			if err = d.TransactionsTable.InsertTransaction(
 				ctx, txn, txnAndSessionID.TransactionID,
@@ -425,7 +455,7 @@ func (d *Database) StoreEvent(
 		return nil
 	})
 	if err != nil {
-		return 0, types.StateAtEvent{}, nil, "", err
+		return 0, types.StateAtEvent{}, nil, "", fmt.Errorf("d.Writer.Do: %w", err)
 	}
 
 	return roomNID, types.StateAtEvent{
@@ -441,7 +471,9 @@ func (d *Database) StoreEvent(
 }
 
 func (d *Database) PublishRoom(ctx context.Context, roomID string, publish bool) error {
-	return d.PublishedTable.UpsertRoomPublished(ctx, roomID, publish)
+	return d.Writer.Do(d.DB, nil, func(_ *sql.Tx) error {
+		return d.PublishedTable.UpsertRoomPublished(ctx, roomID, publish)
+	})
 }
 
 func (d *Database) GetPublishedRooms(ctx context.Context) ([]string, error) {

roomserver/storage/sqlite3/event_json_table.go

@@ -49,15 +49,13 @@ const bulkSelectEventJSONSQL = `
 
 type eventJSONStatements struct {
 	db                      *sql.DB
-	writer                  *sqlutil.TransactionWriter
 	insertEventJSONStmt     *sql.Stmt
 	bulkSelectEventJSONStmt *sql.Stmt
 }
 
 func NewSqliteEventJSONTable(db *sql.DB) (tables.EventJSON, error) {
 	s := &eventJSONStatements{
-		db:     db,
+		db: db,
-		writer: sqlutil.NewTransactionWriter(),
 	}
 	_, err := db.Exec(eventJSONSchema)
 	if err != nil {
@@ -72,10 +70,8 @@ func NewSqliteEventJSONTable(db *sql.DB) (tables.EventJSON, error) {
 func (s *eventJSONStatements) InsertEventJSON(
 	ctx context.Context, txn *sql.Tx, eventNID types.EventNID, eventJSON []byte,
 ) error {
-	return s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	_, err := sqlutil.TxStmt(txn, s.insertEventJSONStmt).ExecContext(ctx, int64(eventNID), eventJSON)
-		_, err := sqlutil.TxStmt(txn, s.insertEventJSONStmt).ExecContext(ctx, int64(eventNID), eventJSON)
+	return err
-		return err
-	})
 }
 
 func (s *eventJSONStatements) BulkSelectEventJSON(

roomserver/storage/sqlite3/event_state_keys_table.go

@@ -64,7 +64,6 @@ const bulkSelectEventStateKeyNIDSQL = `
 
 type eventStateKeyStatements struct {
 	db                             *sql.DB
-	writer                         *sqlutil.TransactionWriter
 	insertEventStateKeyNIDStmt     *sql.Stmt
 	selectEventStateKeyNIDStmt     *sql.Stmt
 	bulkSelectEventStateKeyNIDStmt *sql.Stmt
@@ -73,8 +72,7 @@ type eventStateKeyStatements struct {
 
 func NewSqliteEventStateKeysTable(db *sql.DB) (tables.EventStateKeys, error) {
 	s := &eventStateKeyStatements{
-		db:     db,
+		db: db,
-		writer: sqlutil.NewTransactionWriter(),
 	}
 	_, err := db.Exec(eventStateKeysSchema)
 	if err != nil {
@@ -91,19 +89,15 @@ func NewSqliteEventStateKeysTable(db *sql.DB) (tables.EventStateKeys, error) {
 func (s *eventStateKeyStatements) InsertEventStateKeyNID(
 	ctx context.Context, txn *sql.Tx, eventStateKey string,
 ) (types.EventStateKeyNID, error) {
-	var eventStateKeyNID int64
+	insertStmt := sqlutil.TxStmt(txn, s.insertEventStateKeyNIDStmt)
-	err := s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	res, err := insertStmt.ExecContext(ctx, eventStateKey)
-		insertStmt := sqlutil.TxStmt(txn, s.insertEventStateKeyNIDStmt)
+	if err != nil {
-		res, err := insertStmt.ExecContext(ctx, eventStateKey)
+		return 0, err
-		if err != nil {
+	}
-			return err
+	eventStateKeyNID, err := res.LastInsertId()
-		}
+	if err != nil {
-		eventStateKeyNID, err = res.LastInsertId()
+		return 0, err
-		if err != nil {
+	}
-			return err
-		}
-		return nil
-	})
 	return types.EventStateKeyNID(eventStateKeyNID), err
 }
 

roomserver/storage/sqlite3/event_types_table.go

@@ -18,6 +18,7 @@ package sqlite3
 import (
 	"context"
 	"database/sql"
+	"fmt"
 	"strings"
 
 	"github.com/matrix-org/dendrite/internal"
@@ -78,7 +79,6 @@ const bulkSelectEventTypeNIDSQL = `
 
 type eventTypeStatements struct {
 	db                           *sql.DB
-	writer                       *sqlutil.TransactionWriter
 	insertEventTypeNIDStmt       *sql.Stmt
 	insertEventTypeNIDResultStmt *sql.Stmt
 	selectEventTypeNIDStmt       *sql.Stmt
@@ -87,8 +87,7 @@ type eventTypeStatements struct {
 
 func NewSqliteEventTypesTable(db *sql.DB) (tables.EventTypes, error) {
 	s := &eventTypeStatements{
-		db:     db,
+		db: db,
-		writer: sqlutil.NewTransactionWriter(),
 	}
 	_, err := db.Exec(eventTypesSchema)
 	if err != nil {
@@ -104,18 +103,18 @@ func NewSqliteEventTypesTable(db *sql.DB) (tables.EventTypes, error) {
 }
 
 func (s *eventTypeStatements) InsertEventTypeNID(
-	ctx context.Context, tx *sql.Tx, eventType string,
+	ctx context.Context, txn *sql.Tx, eventType string,
 ) (types.EventTypeNID, error) {
 	var eventTypeNID int64
-	err := s.writer.Do(s.db, tx, func(tx *sql.Tx) error {
+	insertStmt := sqlutil.TxStmt(txn, s.insertEventTypeNIDStmt)
-		insertStmt := sqlutil.TxStmt(tx, s.insertEventTypeNIDStmt)
+	resultStmt := sqlutil.TxStmt(txn, s.insertEventTypeNIDResultStmt)
-		resultStmt := sqlutil.TxStmt(tx, s.insertEventTypeNIDResultStmt)
+	_, err := insertStmt.ExecContext(ctx, eventType)
-		_, err := insertStmt.ExecContext(ctx, eventType)
+	if err != nil {
-		if err != nil {
+		return 0, fmt.Errorf("insertStmt.ExecContext: %w", err)
-			return err
+	}
-		}
+	if err = resultStmt.QueryRowContext(ctx).Scan(&eventTypeNID); err != nil {
-		return resultStmt.QueryRowContext(ctx).Scan(&eventTypeNID)
+		return 0, fmt.Errorf("resultStmt.QueryRowContext.Scan: %w", err)
-	})
+	}
 	return types.EventTypeNID(eventTypeNID), err
 }
 

roomserver/storage/sqlite3/events_table.go

@@ -99,7 +99,6 @@ const selectRoomNIDForEventNIDSQL = "" +
 
 type eventStatements struct {
 	db                                     *sql.DB
-	writer                                 *sqlutil.TransactionWriter
 	insertEventStmt                        *sql.Stmt
 	selectEventStmt                        *sql.Stmt
 	bulkSelectStateEventByIDStmt           *sql.Stmt
@@ -117,8 +116,7 @@ type eventStatements struct {
 
 func NewSqliteEventsTable(db *sql.DB) (tables.Events, error) {
 	s := &eventStatements{
-		db:     db,
+		db: db,
-		writer: sqlutil.NewTransactionWriter(),
 	}
 	_, err := db.Exec(eventsSchema)
 	if err != nil {
@@ -155,22 +153,19 @@ func (s *eventStatements) InsertEvent(
 ) (types.EventNID, types.StateSnapshotNID, error) {
 	// attempt to insert: the last_row_id is the event NID
 	var eventNID int64
-	err := s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	insertStmt := sqlutil.TxStmt(txn, s.insertEventStmt)
-		insertStmt := sqlutil.TxStmt(txn, s.insertEventStmt)
+	result, err := insertStmt.ExecContext(
-		result, err := insertStmt.ExecContext(
+		ctx, int64(roomNID), int64(eventTypeNID), int64(eventStateKeyNID),
-			ctx, int64(roomNID), int64(eventTypeNID), int64(eventStateKeyNID),
+		eventID, referenceSHA256, eventNIDsAsArray(authEventNIDs), depth,
-			eventID, referenceSHA256, eventNIDsAsArray(authEventNIDs), depth,
+	)
-		)
+	if err != nil {
-		if err != nil {
+		return 0, 0, err
-			return err
+	}
-		}
+	modified, err := result.RowsAffected()
-		modified, err := result.RowsAffected()
+	if modified == 0 && err == nil {
-		if modified == 0 && err == nil {
+		return 0, 0, sql.ErrNoRows
-			return sql.ErrNoRows
+	}
-		}
+	eventNID, err = result.LastInsertId()
-		eventNID, err = result.LastInsertId()
-		return err
-	})
 	return types.EventNID(eventNID), 0, err
 }
 
@@ -286,11 +281,8 @@ func (s *eventStatements) BulkSelectStateAtEventByID(
 func (s *eventStatements) UpdateEventState(
 	ctx context.Context, eventNID types.EventNID, stateNID types.StateSnapshotNID,
 ) error {
-	return s.writer.Do(s.db, nil, func(txn *sql.Tx) error {
+	_, err := s.updateEventStateStmt.ExecContext(ctx, int64(stateNID), int64(eventNID))
-		stmt := sqlutil.TxStmt(txn, s.updateEventStateStmt)
+	return err
-		_, err := stmt.ExecContext(ctx, int64(stateNID), int64(eventNID))
-		return err
-	})
 }
 
 func (s *eventStatements) SelectEventSentToOutput(
@@ -302,11 +294,9 @@ func (s *eventStatements) SelectEventSentToOutput(
 }
 
 func (s *eventStatements) UpdateEventSentToOutput(ctx context.Context, txn *sql.Tx, eventNID types.EventNID) error {
-	return s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	updateStmt := sqlutil.TxStmt(txn, s.updateEventSentToOutputStmt)
-		updateStmt := sqlutil.TxStmt(txn, s.updateEventSentToOutputStmt)
+	_, err := updateStmt.ExecContext(ctx, int64(eventNID))
-		_, err := updateStmt.ExecContext(ctx, int64(eventNID))
+	return err
-		return err
-	})
 }
 
 func (s *eventStatements) SelectEventID(
@@ -326,12 +316,16 @@ func (s *eventStatements) BulkSelectStateAtEventAndReference(
 		iEventNIDs[k] = v
 	}
 	selectOrig := strings.Replace(bulkSelectStateAtEventAndReferenceSQL, "($1)", sqlutil.QueryVariadic(len(iEventNIDs)), 1)
-	//////////////
+	selectPrep, err := s.db.Prepare(selectOrig)
-
-	rows, err := txn.QueryContext(ctx, selectOrig, iEventNIDs...)
 	if err != nil {
 		return nil, err
 	}
+	//////////////
+
+	rows, err := sqlutil.TxStmt(txn, selectPrep).QueryContext(ctx, iEventNIDs...)
+	if err != nil {
+		return nil, fmt.Errorf("sqlutil.TxStmt.QueryContext: %w", err)
+	}
 	defer internal.CloseAndLogIfError(ctx, rows, "bulkSelectStateAtEventAndReference: rows.close() failed")
 	results := make([]types.StateAtEventAndReference, len(eventNIDs))
 	i := 0
@@ -372,7 +366,7 @@ func (s *eventStatements) BulkSelectEventReference(
 		iEventNIDs[k] = v
 	}
 	selectOrig := strings.Replace(bulkSelectEventReferenceSQL, "($1)", sqlutil.QueryVariadic(len(iEventNIDs)), 1)
-	selectPrep, err := txn.Prepare(selectOrig)
+	selectPrep, err := s.db.Prepare(selectOrig)
 	if err != nil {
 		return nil, err
 	}
@@ -471,10 +465,14 @@ func (s *eventStatements) SelectMaxEventDepth(ctx context.Context, txn *sql.Tx,
 		iEventIDs[i] = v
 	}
 	sqlStr := strings.Replace(selectMaxEventDepthSQL, "($1)", sqlutil.QueryVariadic(len(iEventIDs)), 1)
-	err := txn.QueryRowContext(ctx, sqlStr, iEventIDs...).Scan(&result)
+	sqlPrep, err := s.db.Prepare(sqlStr)
 	if err != nil {
 		return 0, err
 	}
+	err = sqlutil.TxStmt(txn, sqlPrep).QueryRowContext(ctx, iEventIDs...).Scan(&result)
+	if err != nil {
+		return 0, fmt.Errorf("sqlutil.TxStmt.QueryRowContext: %w", err)
+	}
 	return result, nil
 }
 

roomserver/storage/sqlite3/invite_table.go

@@ -64,7 +64,6 @@ SELECT invite_event_id FROM roomserver_invites WHERE room_nid = $1 AND target_ni
 
 type inviteStatements struct {
 	db                                  *sql.DB
-	writer                              *sqlutil.TransactionWriter
 	insertInviteEventStmt               *sql.Stmt
 	selectInviteActiveForUserInRoomStmt *sql.Stmt
 	updateInviteRetiredStmt             *sql.Stmt
@@ -73,8 +72,7 @@ type inviteStatements struct {
 
 func NewSqliteInvitesTable(db *sql.DB) (tables.Invites, error) {
 	s := &inviteStatements{
-		db:     db,
+		db: db,
-		writer: sqlutil.NewTransactionWriter(),
 	}
 	_, err := db.Exec(inviteSchema)
 	if err != nil {
@@ -96,20 +94,17 @@ func (s *inviteStatements) InsertInviteEvent(
 	inviteEventJSON []byte,
 ) (bool, error) {
 	var count int64
-	err := s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	stmt := sqlutil.TxStmt(txn, s.insertInviteEventStmt)
-		stmt := sqlutil.TxStmt(txn, s.insertInviteEventStmt)
+	result, err := stmt.ExecContext(
-		result, err := stmt.ExecContext(
+		ctx, inviteEventID, roomNID, targetUserNID, senderUserNID, inviteEventJSON,
-			ctx, inviteEventID, roomNID, targetUserNID, senderUserNID, inviteEventJSON,
+	)
-		)
+	if err != nil {
-		if err != nil {
+		return false, err
-			return err
+	}
-		}
+	count, err = result.RowsAffected()
-		count, err = result.RowsAffected()
+	if err != nil {
-		if err != nil {
+		return false, err
-			return err
+	}
-		}
-		return nil
-	})
 	return count != 0, err
 }
 
@@ -117,26 +112,23 @@ func (s *inviteStatements) UpdateInviteRetired(
 	ctx context.Context,
 	txn *sql.Tx, roomNID types.RoomNID, targetUserNID types.EventStateKeyNID,
 ) (eventIDs []string, err error) {
-	err = s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	// gather all the event IDs we will retire
-		// gather all the event IDs we will retire
+	stmt := sqlutil.TxStmt(txn, s.selectInvitesAboutToRetireStmt)
-		stmt := sqlutil.TxStmt(txn, s.selectInvitesAboutToRetireStmt)
+	rows, err := stmt.QueryContext(ctx, roomNID, targetUserNID)
-		rows, err := stmt.QueryContext(ctx, roomNID, targetUserNID)
+	if err != nil {
-		if err != nil {
+		return
-			return err
+	}
+	defer internal.CloseAndLogIfError(ctx, rows, "UpdateInviteRetired: rows.close() failed")
+	for rows.Next() {
+		var inviteEventID string
+		if err = rows.Scan(&inviteEventID); err != nil {
+			return
 		}
-		defer (func() { err = rows.Close() })()
+		eventIDs = append(eventIDs, inviteEventID)
-		for rows.Next() {
+	}
-			var inviteEventID string
+	// now retire the invites
-			if err = rows.Scan(&inviteEventID); err != nil {
+	stmt = sqlutil.TxStmt(txn, s.updateInviteRetiredStmt)
-				return err
+	_, err = stmt.ExecContext(ctx, roomNID, targetUserNID)
-			}
-			eventIDs = append(eventIDs, inviteEventID)
-		}
-		// now retire the invites
-		stmt = sqlutil.TxStmt(txn, s.updateInviteRetiredStmt)
-		_, err = stmt.ExecContext(ctx, roomNID, targetUserNID)
-		return err
-	})
 	return
 }
 

roomserver/storage/sqlite3/membership_table.go

@@ -77,7 +77,6 @@ const updateMembershipSQL = "" +
 
 type membershipStatements struct {
 	db                                              *sql.DB
-	writer                                          *sqlutil.TransactionWriter
 	insertMembershipStmt                            *sql.Stmt
 	selectMembershipForUpdateStmt                   *sql.Stmt
 	selectMembershipFromRoomAndTargetStmt           *sql.Stmt
@@ -90,8 +89,7 @@ type membershipStatements struct {
 
 func NewSqliteMembershipTable(db *sql.DB) (tables.Membership, error) {
 	s := &membershipStatements{
-		db:     db,
+		db: db,
-		writer: sqlutil.NewTransactionWriter(),
 	}
 	_, err := db.Exec(membershipSchema)
 	if err != nil {
@@ -115,11 +113,9 @@ func (s *membershipStatements) InsertMembership(
 	roomNID types.RoomNID, targetUserNID types.EventStateKeyNID,
 	localTarget bool,
 ) error {
-	return s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	stmt := sqlutil.TxStmt(txn, s.insertMembershipStmt)
-		stmt := sqlutil.TxStmt(txn, s.insertMembershipStmt)
+	_, err := stmt.ExecContext(ctx, roomNID, targetUserNID, localTarget)
-		_, err := stmt.ExecContext(ctx, roomNID, targetUserNID, localTarget)
+	return err
-		return err
-	})
 }
 
 func (s *membershipStatements) SelectMembershipForUpdate(
@@ -201,11 +197,9 @@ func (s *membershipStatements) UpdateMembership(
 	senderUserNID types.EventStateKeyNID, membership tables.MembershipState,
 	eventNID types.EventNID,
 ) error {
-	return s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	stmt := sqlutil.TxStmt(txn, s.updateMembershipStmt)
-		stmt := sqlutil.TxStmt(txn, s.updateMembershipStmt)
+	_, err := stmt.ExecContext(
-		_, err := stmt.ExecContext(
+		ctx, senderUserNID, membership, eventNID, roomNID, targetUserNID,
-			ctx, senderUserNID, membership, eventNID, roomNID, targetUserNID,
+	)
-		)
+	return err
-		return err
-	})
 }

roomserver/storage/sqlite3/previous_events_table.go

@@ -54,15 +54,13 @@ const selectPreviousEventExistsSQL = `
 
 type previousEventStatements struct {
 	db                            *sql.DB
-	writer                        *sqlutil.TransactionWriter
 	insertPreviousEventStmt       *sql.Stmt
 	selectPreviousEventExistsStmt *sql.Stmt
 }
 
 func NewSqlitePrevEventsTable(db *sql.DB) (tables.PreviousEvents, error) {
 	s := &previousEventStatements{
-		db:     db,
+		db: db,
-		writer: sqlutil.NewTransactionWriter(),
 	}
 	_, err := db.Exec(previousEventSchema)
 	if err != nil {
@@ -82,13 +80,11 @@ func (s *previousEventStatements) InsertPreviousEvent(
 	previousEventReferenceSHA256 []byte,
 	eventNID types.EventNID,
 ) error {
-	return s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	stmt := sqlutil.TxStmt(txn, s.insertPreviousEventStmt)
-		stmt := sqlutil.TxStmt(txn, s.insertPreviousEventStmt)
+	_, err := stmt.ExecContext(
-		_, err := stmt.ExecContext(
+		ctx, previousEventID, previousEventReferenceSHA256, int64(eventNID),
-			ctx, previousEventID, previousEventReferenceSHA256, int64(eventNID),
+	)
-		)
+	return err
-		return err
-	})
 }
 
 // Check if the event reference exists

roomserver/storage/sqlite3/published_table.go

@@ -19,7 +19,6 @@ import (
 	"database/sql"
 
 	"github.com/matrix-org/dendrite/internal"
-	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/roomserver/storage/shared"
 	"github.com/matrix-org/dendrite/roomserver/storage/tables"
 )
@@ -45,7 +44,6 @@ const selectPublishedSQL = "" +
 
 type publishedStatements struct {
 	db                     *sql.DB
-	writer                 *sqlutil.TransactionWriter
 	upsertPublishedStmt    *sql.Stmt
 	selectAllPublishedStmt *sql.Stmt
 	selectPublishedStmt    *sql.Stmt
@@ -53,8 +51,7 @@ type publishedStatements struct {
 
 func NewSqlitePublishedTable(db *sql.DB) (tables.Published, error) {
 	s := &publishedStatements{
-		db:     db,
+		db: db,
-		writer: sqlutil.NewTransactionWriter(),
 	}
 	_, err := db.Exec(publishedSchema)
 	if err != nil {
@@ -69,12 +66,9 @@ func NewSqlitePublishedTable(db *sql.DB) (tables.Published, error) {
 
 func (s *publishedStatements) UpsertRoomPublished(
 	ctx context.Context, roomID string, published bool,
-) (err error) {
+) error {
-	return s.writer.Do(s.db, nil, func(txn *sql.Tx) error {
+	_, err := s.upsertPublishedStmt.ExecContext(ctx, roomID, published)
-		stmt := sqlutil.TxStmt(txn, s.upsertPublishedStmt)
+	return err
-		_, err := stmt.ExecContext(ctx, roomID, published)
-		return err
-	})
 }
 
 func (s *publishedStatements) SelectPublishedFromRoomID(

roomserver/storage/sqlite3/redactions_table.go

@@ -53,7 +53,6 @@ const markRedactionValidatedSQL = "" +
 
 type redactionStatements struct {
 	db                                          *sql.DB
-	writer                                      *sqlutil.TransactionWriter
 	insertRedactionStmt                         *sql.Stmt
 	selectRedactionInfoByRedactionEventIDStmt   *sql.Stmt
 	selectRedactionInfoByEventBeingRedactedStmt *sql.Stmt
@@ -62,8 +61,7 @@ type redactionStatements struct {
 
 func NewSqliteRedactionsTable(db *sql.DB) (tables.Redactions, error) {
 	s := &redactionStatements{
-		db:     db,
+		db: db,
-		writer: sqlutil.NewTransactionWriter(),
 	}
 	_, err := db.Exec(redactionsSchema)
 	if err != nil {
@@ -81,11 +79,9 @@ func NewSqliteRedactionsTable(db *sql.DB) (tables.Redactions, error) {
 func (s *redactionStatements) InsertRedaction(
 	ctx context.Context, txn *sql.Tx, info tables.RedactionInfo,
 ) error {
-	return s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	stmt := sqlutil.TxStmt(txn, s.insertRedactionStmt)
-		stmt := sqlutil.TxStmt(txn, s.insertRedactionStmt)
+	_, err := stmt.ExecContext(ctx, info.RedactionEventID, info.RedactsEventID, info.Validated)
-		_, err := stmt.ExecContext(ctx, info.RedactionEventID, info.RedactsEventID, info.Validated)
+	return err
-		return err
-	})
 }
 
 func (s *redactionStatements) SelectRedactionInfoByRedactionEventID(
@@ -121,9 +117,7 @@ func (s *redactionStatements) SelectRedactionInfoByEventBeingRedacted(
 func (s *redactionStatements) MarkRedactionValidated(
 	ctx context.Context, txn *sql.Tx, redactionEventID string, validated bool,
 ) error {
-	return s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	stmt := sqlutil.TxStmt(txn, s.markRedactionValidatedStmt)
-		stmt := sqlutil.TxStmt(txn, s.markRedactionValidatedStmt)
+	_, err := stmt.ExecContext(ctx, redactionEventID, validated)
-		_, err := stmt.ExecContext(ctx, redactionEventID, validated)
+	return err
-		return err
-	})
 }

roomserver/storage/sqlite3/room_aliases_table.go

@@ -20,7 +20,6 @@ import (
 	"database/sql"
 
 	"github.com/matrix-org/dendrite/internal"
-	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/roomserver/storage/shared"
 	"github.com/matrix-org/dendrite/roomserver/storage/tables"
 )
@@ -57,7 +56,6 @@ const deleteRoomAliasSQL = `
 
 type roomAliasesStatements struct {
 	db                           *sql.DB
-	writer                       *sqlutil.TransactionWriter
 	insertRoomAliasStmt          *sql.Stmt
 	selectRoomIDFromAliasStmt    *sql.Stmt
 	selectAliasesFromRoomIDStmt  *sql.Stmt
@@ -67,8 +65,7 @@ type roomAliasesStatements struct {
 
 func NewSqliteRoomAliasesTable(db *sql.DB) (tables.RoomAliases, error) {
 	s := &roomAliasesStatements{
-		db:     db,
+		db: db,
-		writer: sqlutil.NewTransactionWriter(),
 	}
 	_, err := db.Exec(roomAliasesSchema)
 	if err != nil {
@@ -85,12 +82,9 @@ func NewSqliteRoomAliasesTable(db *sql.DB) (tables.RoomAliases, error) {
 
 func (s *roomAliasesStatements) InsertRoomAlias(
 	ctx context.Context, alias string, roomID string, creatorUserID string,
-) (err error) {
+) error {
-	return s.writer.Do(s.db, nil, func(txn *sql.Tx) error {
+	_, err := s.insertRoomAliasStmt.ExecContext(ctx, alias, roomID, creatorUserID)
-		stmt := sqlutil.TxStmt(txn, s.insertRoomAliasStmt)
+	return err
-		_, err := stmt.ExecContext(ctx, alias, roomID, creatorUserID)
-		return err
-	})
 }
 
 func (s *roomAliasesStatements) SelectRoomIDFromAlias(
@@ -138,10 +132,7 @@ func (s *roomAliasesStatements) SelectCreatorIDFromAlias(
 
 func (s *roomAliasesStatements) DeleteRoomAlias(
 	ctx context.Context, alias string,
-) (err error) {
+) error {
-	return s.writer.Do(s.db, nil, func(txn *sql.Tx) error {
+	_, err := s.deleteRoomAliasStmt.ExecContext(ctx, alias)
-		stmt := sqlutil.TxStmt(txn, s.deleteRoomAliasStmt)
+	return err
-		_, err := stmt.ExecContext(ctx, alias)
-		return err
-	})
 }

roomserver/storage/sqlite3/rooms_table.go

@@ -66,7 +66,6 @@ const selectRoomVersionForRoomNIDSQL = "" +
 
 type roomStatements struct {
 	db                                 *sql.DB
-	writer                             *sqlutil.TransactionWriter
 	insertRoomNIDStmt                  *sql.Stmt
 	selectRoomNIDStmt                  *sql.Stmt
 	selectLatestEventNIDsStmt          *sql.Stmt
@@ -78,8 +77,7 @@ type roomStatements struct {
 
 func NewSqliteRoomsTable(db *sql.DB) (tables.Rooms, error) {
 	s := &roomStatements{
-		db:     db,
+		db: db,
-		writer: sqlutil.NewTransactionWriter(),
 	}
 	_, err := db.Exec(roomsSchema)
 	if err != nil {
@@ -100,20 +98,14 @@ func (s *roomStatements) InsertRoomNID(
 	ctx context.Context, txn *sql.Tx,
 	roomID string, roomVersion gomatrixserverlib.RoomVersion,
 ) (roomNID types.RoomNID, err error) {
-	err = s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	insertStmt := sqlutil.TxStmt(txn, s.insertRoomNIDStmt)
-		insertStmt := sqlutil.TxStmt(txn, s.insertRoomNIDStmt)
+	_, err = insertStmt.ExecContext(ctx, roomID, roomVersion)
-		_, err = insertStmt.ExecContext(ctx, roomID, roomVersion)
-		if err != nil {
-			return fmt.Errorf("insertStmt.ExecContext: %w", err)
-		}
-		roomNID, err = s.SelectRoomNID(ctx, txn, roomID)
-		if err != nil {
-			return fmt.Errorf("s.SelectRoomNID: %w", err)
-		}
-		return nil
-	})
 	if err != nil {
-		return types.RoomNID(0), err
+		return 0, fmt.Errorf("insertStmt.ExecContext: %w", err)
+	}
+	roomNID, err = s.SelectRoomNID(ctx, txn, roomID)
+	if err != nil {
+		return 0, fmt.Errorf("s.SelectRoomNID: %w", err)
 	}
 	return
 }
@@ -170,17 +162,15 @@ func (s *roomStatements) UpdateLatestEventNIDs(
 	lastEventSentNID types.EventNID,
 	stateSnapshotNID types.StateSnapshotNID,
 ) error {
-	return s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	stmt := sqlutil.TxStmt(txn, s.updateLatestEventNIDsStmt)
-		stmt := sqlutil.TxStmt(txn, s.updateLatestEventNIDsStmt)
+	_, err := stmt.ExecContext(
-		_, err := stmt.ExecContext(
+		ctx,
-			ctx,
+		eventNIDsAsArray(eventNIDs),
-			eventNIDsAsArray(eventNIDs),
+		int64(lastEventSentNID),
-			int64(lastEventSentNID),
+		int64(stateSnapshotNID),
-			int64(stateSnapshotNID),
+		roomNID,
-			roomNID,
+	)
-		)
+	return err
-		return err
-	})
 }
 
 func (s *roomStatements) SelectRoomVersionForRoomID(

roomserver/storage/sqlite3/state_block_table.go

@@ -74,7 +74,6 @@ const bulkSelectFilteredStateBlockEntriesSQL = "" +
 
 type stateBlockStatements struct {
 	db                                      *sql.DB
-	writer                                  *sqlutil.TransactionWriter
 	insertStateDataStmt                     *sql.Stmt
 	selectNextStateBlockNIDStmt             *sql.Stmt
 	bulkSelectStateBlockEntriesStmt         *sql.Stmt
@@ -83,8 +82,7 @@ type stateBlockStatements struct {
 
 func NewSqliteStateBlockTable(db *sql.DB) (tables.StateBlock, error) {
 	s := &stateBlockStatements{
-		db:     db,
+		db: db,
-		writer: sqlutil.NewTransactionWriter(),
 	}
 	_, err := db.Exec(stateDataSchema)
 	if err != nil {
@@ -107,25 +105,22 @@ func (s *stateBlockStatements) BulkInsertStateData(
 		return 0, nil
 	}
 	var stateBlockNID types.StateBlockNID
-	err := s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	err := txn.Stmt(s.selectNextStateBlockNIDStmt).QueryRowContext(ctx).Scan(&stateBlockNID)
-		err := txn.Stmt(s.selectNextStateBlockNIDStmt).QueryRowContext(ctx).Scan(&stateBlockNID)
+	if err != nil {
+		return 0, err
+	}
+	for _, entry := range entries {
+		_, err = txn.Stmt(s.insertStateDataStmt).ExecContext(
+			ctx,
+			int64(stateBlockNID),
+			int64(entry.EventTypeNID),
+			int64(entry.EventStateKeyNID),
+			int64(entry.EventNID),
+		)
 		if err != nil {
-			return err
+			return 0, err
 		}
-		for _, entry := range entries {
+	}
-			_, err := txn.Stmt(s.insertStateDataStmt).ExecContext(
-				ctx,
-				int64(stateBlockNID),
-				int64(entry.EventTypeNID),
-				int64(entry.EventStateKeyNID),
-				int64(entry.EventNID),
-			)
-			if err != nil {
-				return err
-			}
-		}
-		return nil
-	})
 	return stateBlockNID, err
 }
 

roomserver/storage/sqlite3/state_snapshot_table.go

@@ -50,15 +50,13 @@ const bulkSelectStateBlockNIDsSQL = "" +
 
 type stateSnapshotStatements struct {
 	db                           *sql.DB
-	writer                       *sqlutil.TransactionWriter
 	insertStateStmt              *sql.Stmt
 	bulkSelectStateBlockNIDsStmt *sql.Stmt
 }
 
 func NewSqliteStateSnapshotTable(db *sql.DB) (tables.StateSnapshot, error) {
 	s := &stateSnapshotStatements{
-		db:     db,
+		db: db,
-		writer: sqlutil.NewTransactionWriter(),
 	}
 	_, err := db.Exec(stateSnapshotSchema)
 	if err != nil {
@@ -78,19 +76,16 @@ func (s *stateSnapshotStatements) InsertState(
 	if err != nil {
 		return
 	}
-	err = s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	insertStmt := txn.Stmt(s.insertStateStmt)
-		insertStmt := txn.Stmt(s.insertStateStmt)
+	res, err := insertStmt.ExecContext(ctx, int64(roomNID), string(stateBlockNIDsJSON))
-		res, err := insertStmt.ExecContext(ctx, int64(roomNID), string(stateBlockNIDsJSON))
+	if err != nil {
-		if err != nil {
+		return 0, err
-			return err
+	}
-		}
+	lastRowID, err := res.LastInsertId()
-		lastRowID, err := res.LastInsertId()
+	if err != nil {
-		if err != nil {
+		return 0, err
-			return err
+	}
-		}
+	stateNID = types.StateSnapshotNID(lastRowID)
-		stateNID = types.StateSnapshotNID(lastRowID)
-		return nil
-	})
 	return
 }
 

roomserver/storage/sqlite3/storage.go

@@ -41,6 +41,7 @@ type Database struct {
 	invites        tables.Invites
 	membership     tables.Membership
 	db             *sql.DB
+	writer         sqlutil.TransactionWriter
 }
 
 // Open a sqlite database.
@@ -51,6 +52,7 @@ func Open(dbProperties *config.DatabaseOptions) (*Database, error) {
 	if d.db, err = sqlutil.Open(dbProperties); err != nil {
 		return nil, err
 	}
+	d.writer = sqlutil.NewTransactionWriter()
 	//d.db.Exec("PRAGMA journal_mode=WAL;")
 	//d.db.Exec("PRAGMA read_uncommitted = true;")
 
@@ -118,6 +120,7 @@ func Open(dbProperties *config.DatabaseOptions) (*Database, error) {
 	}
 	d.Database = shared.Database{
 		DB:                  d.db,
+		Writer:              sqlutil.NewTransactionWriter(),
 		EventsTable:         d.events,
 		EventTypesTable:     d.eventTypes,
 		EventStateKeysTable: d.eventStateKeys,
@@ -138,25 +141,25 @@ func Open(dbProperties *config.DatabaseOptions) (*Database, error) {
 
 func (d *Database) GetLatestEventsForUpdate(
 	ctx context.Context, roomNID types.RoomNID,
-) (types.RoomRecentEventsUpdater, error) {
+) (*shared.LatestEventsUpdater, error) {
 	// TODO: Do not use transactions. 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. As sqlite doesn't support multi-process on the
 	// same DB anyway, and we only execute updates sequentially, the only worries
 	// are for rolling back when things go wrong. (atomicity)
-	return shared.NewRoomRecentEventsUpdater(&d.Database, ctx, roomNID, false)
+	return shared.NewLatestEventsUpdater(ctx, &d.Database, nil, roomNID)
 }
 
 func (d *Database) MembershipUpdater(
 	ctx context.Context, roomID, targetUserID string,
 	targetLocal bool, roomVersion gomatrixserverlib.RoomVersion,
-) (updater types.MembershipUpdater, err error) {
+) (*shared.MembershipUpdater, error) {
 	// TODO: Do not use transactions. 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. As sqlite doesn't support multi-process on the
 	// same DB anyway, and we only execute updates sequentially, the only worries
 	// are for rolling back when things go wrong. (atomicity)
-	return shared.NewMembershipUpdater(ctx, &d.Database, roomID, targetUserID, targetLocal, roomVersion, false)
+	return shared.NewMembershipUpdater(ctx, &d.Database, nil, roomID, targetUserID, targetLocal, roomVersion)
 }

roomserver/storage/sqlite3/transactions_table.go

@@ -45,15 +45,13 @@ const selectTransactionEventIDSQL = `
 
 type transactionStatements struct {
 	db                           *sql.DB
-	writer                       *sqlutil.TransactionWriter
 	insertTransactionStmt        *sql.Stmt
 	selectTransactionEventIDStmt *sql.Stmt
 }
 
 func NewSqliteTransactionsTable(db *sql.DB) (tables.Transactions, error) {
 	s := &transactionStatements{
-		db:     db,
+		db: db,
-		writer: sqlutil.NewTransactionWriter(),
 	}
 	_, err := db.Exec(transactionsSchema)
 	if err != nil {
@@ -72,14 +70,12 @@ func (s *transactionStatements) InsertTransaction(
 	sessionID int64,
 	userID string,
 	eventID string,
-) (err error) {
+) error {
-	return s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+	stmt := sqlutil.TxStmt(txn, s.insertTransactionStmt)
-		stmt := sqlutil.TxStmt(txn, s.insertTransactionStmt)
+	_, err := stmt.ExecContext(
-		_, err := stmt.ExecContext(
+		ctx, transactionID, sessionID, userID, eventID,
-			ctx, transactionID, sessionID, userID, eventID,
+	)
-		)
+	return err
-		return err
-	})
 }
 
 func (s *transactionStatements) SelectTransactionEventID(

roomserver/types/types.go

@@ -16,7 +16,6 @@
 package types
 
 import (
-	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/gomatrixserverlib"
 )
 
@@ -140,68 +139,6 @@ type StateEntryList struct {
 	StateEntries  []StateEntry
 }
 
-// A RoomRecentEventsUpdater is used to update the recent events in a room.
-// (On postgresql this wraps a database transaction that holds a "FOR UPDATE"
-//  lock on the row in the rooms table holding the latest events for the room.)
-type RoomRecentEventsUpdater interface {
-	// The room version of the room.
-	RoomVersion() gomatrixserverlib.RoomVersion
-	// The latest event IDs and state in the room.
-	LatestEvents() []StateAtEventAndReference
-	// The event ID of the latest event written to the output log in the room.
-	LastEventIDSent() string
-	// The current state of the room.
-	CurrentStateSnapshotNID() StateSnapshotNID
-	// Store the previous events referenced by an event.
-	// This adds the event NID to an entry in the database for each of the previous events.
-	// If there isn't an entry for one of previous events then an entry is created.
-	// If the entry already lists the event NID as a referrer then the entry unmodified.
-	// (i.e. the operation is idempotent)
-	StorePreviousEvents(eventNID EventNID, previousEventReferences []gomatrixserverlib.EventReference) error
-	// Check whether the eventReference is already referenced by another matrix event.
-	IsReferenced(eventReference gomatrixserverlib.EventReference) (bool, error)
-	// Set the list of latest events for the room.
-	// This replaces the current list stored in the database with the given list
-	SetLatestEvents(
-		roomNID RoomNID, latest []StateAtEventAndReference, lastEventNIDSent EventNID,
-		currentStateSnapshotNID StateSnapshotNID,
-	) error
-	// Check if the event has already be written to the output logs.
-	HasEventBeenSent(eventNID EventNID) (bool, error)
-	// Mark the event as having been sent to the output logs.
-	MarkEventAsSent(eventNID EventNID) error
-	// Build a membership updater for the target user in this room.
-	// It will share the same transaction as this updater.
-	MembershipUpdater(targetUserNID EventStateKeyNID, isTargetLocalUser bool) (MembershipUpdater, error)
-	// Implements Transaction so it can be committed or rolledback
-	sqlutil.Transaction
-}
-
-// A MembershipUpdater is used to update the membership of a user in a room.
-// (On postgresql this wraps a database transaction that holds a "FOR UPDATE"
-//  lock on the row in the membership table for this user in the room)
-// The caller should call one of SetToInvite, SetToJoin or SetToLeave once to
-// make the update, or none of them if no update is required.
-type MembershipUpdater interface {
-	// True if the target user is invited to the room before updating.
-	IsInvite() bool
-	// True if the target user is joined to the room before updating.
-	IsJoin() bool
-	// True if the target user is not invited or joined to the room before updating.
-	IsLeave() bool
-	// Set the state to invite.
-	// Returns whether this invite needs to be sent
-	SetToInvite(event gomatrixserverlib.Event) (needsSending bool, err error)
-	// Set the state to join or updates the event ID in the database.
-	// Returns a list of invite event IDs that this state change retired.
-	SetToJoin(senderUserID string, eventID string, isUpdate bool) (inviteEventIDs []string, err error)
-	// Set the state to leave.
-	// Returns a list of invite event IDs that this state change retired.
-	SetToLeave(senderUserID string, eventID string) (inviteEventIDs []string, err error)
-	// Implements Transaction so it can be committed or rolledback.
-	sqlutil.Transaction
-}
-
 // A MissingEventError is an error that happened because the roomserver was
 // missing requested events from its database.
 type MissingEventError string

serverkeyapi/storage/sqlite3/server_key_table.go

@@ -63,7 +63,7 @@ const upsertServerKeysSQL = "" +
 
 type serverKeyStatements struct {
 	db                       *sql.DB
-	writer                   *sqlutil.TransactionWriter
+	writer                   sqlutil.TransactionWriter
 	bulkSelectServerKeysStmt *sql.Stmt
 	upsertServerKeysStmt     *sql.Stmt
 }

syncapi/storage/shared/syncserver.go

@@ -45,7 +45,7 @@ type Database struct {
 	BackwardExtremities tables.BackwardsExtremities
 	SendToDevice        tables.SendToDevice
 	Filter              tables.Filter
-	SendToDeviceWriter  *sqlutil.TransactionWriter
+	SendToDeviceWriter  sqlutil.TransactionWriter
 	EDUCache            *cache.EDUCache
 }
 

syncapi/storage/sqlite3/account_data_table.go

@@ -51,7 +51,7 @@ const selectMaxAccountDataIDSQL = "" +
 
 type accountDataStatements struct {
 	db                           *sql.DB
-	writer                       *sqlutil.TransactionWriter
+	writer                       sqlutil.TransactionWriter
 	streamIDStatements           *streamIDStatements
 	insertAccountDataStmt        *sql.Stmt
 	selectMaxAccountDataIDStmt   *sql.Stmt

syncapi/storage/sqlite3/backwards_extremities_table.go

@@ -49,7 +49,7 @@ const deleteBackwardExtremitySQL = "" +
 
 type backwardExtremitiesStatements struct {
 	db                                   *sql.DB
-	writer                               *sqlutil.TransactionWriter
+	writer                               sqlutil.TransactionWriter
 	insertBackwardExtremityStmt          *sql.Stmt
 	selectBackwardExtremitiesForRoomStmt *sql.Stmt
 	deleteBackwardExtremityStmt          *sql.Stmt

syncapi/storage/sqlite3/current_room_state_table.go

@@ -85,7 +85,7 @@ const selectEventsWithEventIDsSQL = "" +
 
 type currentRoomStateStatements struct {
 	db                              *sql.DB
-	writer                          *sqlutil.TransactionWriter
+	writer                          sqlutil.TransactionWriter
 	streamIDStatements              *streamIDStatements
 	upsertRoomStateStmt             *sql.Stmt
 	deleteRoomStateByEventIDStmt    *sql.Stmt

syncapi/storage/sqlite3/filter_table.go

@@ -52,7 +52,7 @@ const insertFilterSQL = "" +
 
 type filterStatements struct {
 	db                          *sql.DB
-	writer                      *sqlutil.TransactionWriter
+	writer                      sqlutil.TransactionWriter
 	selectFilterStmt            *sql.Stmt
 	selectFilterIDByContentStmt *sql.Stmt
 	insertFilterStmt            *sql.Stmt

syncapi/storage/sqlite3/invites_table.go

@@ -59,7 +59,7 @@ const selectMaxInviteIDSQL = "" +
 
 type inviteEventsStatements struct {
 	db                            *sql.DB
-	writer                        *sqlutil.TransactionWriter
+	writer                        sqlutil.TransactionWriter
 	streamIDStatements            *streamIDStatements
 	insertInviteEventStmt         *sql.Stmt
 	selectInviteEventsInRangeStmt *sql.Stmt

syncapi/storage/sqlite3/output_room_events_table.go

@@ -105,7 +105,7 @@ const selectStateInRangeSQL = "" +
 
 type outputRoomEventsStatements struct {
 	db                            *sql.DB
-	writer                        *sqlutil.TransactionWriter
+	writer                        sqlutil.TransactionWriter
 	streamIDStatements            *streamIDStatements
 	insertEventStmt               *sql.Stmt
 	selectEventsStmt              *sql.Stmt

syncapi/storage/sqlite3/output_room_events_topology_table.go

@@ -67,7 +67,7 @@ const selectMaxPositionInTopologySQL = "" +
 
 type outputRoomEventsTopologyStatements struct {
 	db                              *sql.DB
-	writer                          *sqlutil.TransactionWriter
+	writer                          sqlutil.TransactionWriter
 	insertEventInTopologyStmt       *sql.Stmt
 	selectEventIDsInRangeASCStmt    *sql.Stmt
 	selectEventIDsInRangeDESCStmt   *sql.Stmt

syncapi/storage/sqlite3/send_to_device_table.go

@@ -73,7 +73,7 @@ const deleteSendToDeviceMessagesSQL = `
 
 type sendToDeviceStatements struct {
 	db                             *sql.DB
-	writer                         *sqlutil.TransactionWriter
+	writer                         sqlutil.TransactionWriter
 	insertSendToDeviceMessageStmt  *sql.Stmt
 	selectSendToDeviceMessagesStmt *sql.Stmt
 	countSendToDeviceMessagesStmt  *sql.Stmt

syncapi/storage/sqlite3/stream_id_table.go

@@ -28,7 +28,7 @@ const selectStreamIDStmt = "" +
 
 type streamIDStatements struct {
 	db                   *sql.DB
-	writer               *sqlutil.TransactionWriter
+	writer               sqlutil.TransactionWriter
 	increaseStreamIDStmt *sql.Stmt
 	selectStreamIDStmt   *sql.Stmt
 }

userapi/storage/accounts/sqlite3/account_data_table.go

@@ -51,7 +51,7 @@ const selectAccountDataByTypeSQL = "" +
 
 type accountDataStatements struct {
 	db                          *sql.DB
-	writer                      *sqlutil.TransactionWriter
+	writer                      sqlutil.TransactionWriter
 	insertAccountDataStmt       *sql.Stmt
 	selectAccountDataStmt       *sql.Stmt
 	selectAccountDataByTypeStmt *sql.Stmt

userapi/storage/accounts/sqlite3/accounts_table.go

@@ -59,7 +59,7 @@ const selectNewNumericLocalpartSQL = "" +
 
 type accountsStatements struct {
 	db                            *sql.DB
-	writer                        *sqlutil.TransactionWriter
+	writer                        sqlutil.TransactionWriter
 	insertAccountStmt             *sql.Stmt
 	selectAccountByLocalpartStmt  *sql.Stmt
 	selectPasswordHashStmt        *sql.Stmt

userapi/storage/accounts/sqlite3/profile_table.go

@@ -53,7 +53,7 @@ const selectProfilesBySearchSQL = "" +
 
 type profilesStatements struct {
 	db                           *sql.DB
-	writer                       *sqlutil.TransactionWriter
+	writer                       sqlutil.TransactionWriter
 	insertProfileStmt            *sql.Stmt
 	selectProfileByLocalpartStmt *sql.Stmt
 	setAvatarURLStmt             *sql.Stmt

userapi/storage/accounts/sqlite3/threepid_table.go

@@ -54,7 +54,7 @@ const deleteThreePIDSQL = "" +
 
 type threepidStatements struct {
 	db                              *sql.DB
-	writer                          *sqlutil.TransactionWriter
+	writer                          sqlutil.TransactionWriter
 	selectLocalpartForThreePIDStmt  *sql.Stmt
 	selectThreePIDsForLocalpartStmt *sql.Stmt
 	insertThreePIDStmt              *sql.Stmt

userapi/storage/devices/sqlite3/devices_table.go

@@ -78,7 +78,7 @@ const selectDevicesByIDSQL = "" +
 
 type devicesStatements struct {
 	db                           *sql.DB
-	writer                       *sqlutil.TransactionWriter
+	writer                       sqlutil.TransactionWriter
 	insertDeviceStmt             *sql.Stmt
 	selectDevicesCountStmt       *sql.Stmt
 	selectDeviceByTokenStmt      *sql.Stmt