Consistent `*sql.Tx` usage across sync API (#2744)

This tidies up the `storage` package so that everything takes a
transaction parameter instead of something things that do and some that
don't.

20 files changed, 99 insertions, 77 deletions

diff --git a/syncapi/storage/postgres/account_data_table.go b/syncapi/storage/postgres/account_data_table.go
index e9c72058..aa54cb08 100644
--- a/syncapi/storage/postgres/account_data_table.go
+++ b/syncapi/storage/postgres/account_data_table.go
@@ -99,14 +99,15 @@ func (s *accountDataStatements) InsertAccountData(
 }
 
 func (s *accountDataStatements) SelectAccountDataInRange(
-	ctx context.Context,
+	ctx context.Context, txn *sql.Tx,
 	userID string,
 	r types.Range,
 	accountDataEventFilter *gomatrixserverlib.EventFilter,
 ) (data map[string][]string, pos types.StreamPosition, err error) {
 	data = make(map[string][]string)
 
-	rows, err := s.selectAccountDataInRangeStmt.QueryContext(ctx, userID, r.Low(), r.High(),
+	rows, err := sqlutil.TxStmt(txn, s.selectAccountDataInRangeStmt).QueryContext(
+		ctx, userID, r.Low(), r.High(),
 		pq.StringArray(filterConvertTypeWildcardToSQL(accountDataEventFilter.Types)),
 		pq.StringArray(filterConvertTypeWildcardToSQL(accountDataEventFilter.NotTypes)),
 		accountDataEventFilter.Limit,
diff --git a/syncapi/storage/postgres/backwards_extremities_table.go b/syncapi/storage/postgres/backwards_extremities_table.go
index d4515735..8fc92091 100644
--- a/syncapi/storage/postgres/backwards_extremities_table.go
+++ b/syncapi/storage/postgres/backwards_extremities_table.go
@@ -79,9 +79,9 @@ func (s *backwardExtremitiesStatements) InsertsBackwardExtremity(
 }
 
 func (s *backwardExtremitiesStatements) SelectBackwardExtremitiesForRoom(
-	ctx context.Context, roomID string,
+	ctx context.Context, txn *sql.Tx, roomID string,
 ) (bwExtrems map[string][]string, err error) {
-	rows, err := s.selectBackwardExtremitiesForRoomStmt.QueryContext(ctx, roomID)
+	rows, err := sqlutil.TxStmt(txn, s.selectBackwardExtremitiesForRoomStmt).QueryContext(ctx, roomID)
 	if err != nil {
 		return
 	}
diff --git a/syncapi/storage/postgres/current_room_state_table.go b/syncapi/storage/postgres/current_room_state_table.go
index 5e6daaaf..4ffd2961 100644
--- a/syncapi/storage/postgres/current_room_state_table.go
+++ b/syncapi/storage/postgres/current_room_state_table.go
@@ -185,9 +185,9 @@ func NewPostgresCurrentRoomStateTable(db *sql.DB) (tables.CurrentRoomState, erro
 
 // SelectJoinedUsers returns a map of room ID to a list of joined user IDs.
 func (s *currentRoomStateStatements) SelectJoinedUsers(
-	ctx context.Context,
+	ctx context.Context, txn *sql.Tx,
 ) (map[string][]string, error) {
-	rows, err := s.selectJoinedUsersStmt.QueryContext(ctx)
+	rows, err := sqlutil.TxStmt(txn, s.selectJoinedUsersStmt).QueryContext(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -209,9 +209,9 @@ func (s *currentRoomStateStatements) SelectJoinedUsers(
 
 // SelectJoinedUsersInRoom returns a map of room ID to a list of joined user IDs for a given room.
 func (s *currentRoomStateStatements) SelectJoinedUsersInRoom(
-	ctx context.Context, roomIDs []string,
+	ctx context.Context, txn *sql.Tx, roomIDs []string,
 ) (map[string][]string, error) {
-	rows, err := s.selectJoinedUsersInRoomStmt.QueryContext(ctx, pq.StringArray(roomIDs))
+	rows, err := sqlutil.TxStmt(txn, s.selectJoinedUsersInRoomStmt).QueryContext(ctx, pq.StringArray(roomIDs))
 	if err != nil {
 		return nil, err
 	}
@@ -387,9 +387,9 @@ func rowsToEvents(rows *sql.Rows) ([]*gomatrixserverlib.HeaderedEvent, error) {
 }
 
 func (s *currentRoomStateStatements) SelectStateEvent(
-	ctx context.Context, roomID, evType, stateKey string,
+	ctx context.Context, txn *sql.Tx, roomID, evType, stateKey string,
 ) (*gomatrixserverlib.HeaderedEvent, error) {
-	stmt := s.selectStateEventStmt
+	stmt := sqlutil.TxStmt(txn, s.selectStateEventStmt)
 	var res []byte
 	err := stmt.QueryRowContext(ctx, roomID, evType, stateKey).Scan(&res)
 	if err == sql.ErrNoRows {
diff --git a/syncapi/storage/postgres/filter_table.go b/syncapi/storage/postgres/filter_table.go
index c82ef092..86cec362 100644
--- a/syncapi/storage/postgres/filter_table.go
+++ b/syncapi/storage/postgres/filter_table.go
@@ -19,6 +19,7 @@ import (
 	"database/sql"
 	"encoding/json"
 
+	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
 	"github.com/matrix-org/gomatrixserverlib"
 )
@@ -73,11 +74,11 @@ func NewPostgresFilterTable(db *sql.DB) (tables.Filter, error) {
 }
 
 func (s *filterStatements) SelectFilter(
-	ctx context.Context, target *gomatrixserverlib.Filter, localpart string, filterID string,
+	ctx context.Context, txn *sql.Tx, target *gomatrixserverlib.Filter, localpart string, filterID string,
 ) error {
 	// Retrieve filter from database (stored as canonical JSON)
 	var filterData []byte
-	err := s.selectFilterStmt.QueryRowContext(ctx, localpart, filterID).Scan(&filterData)
+	err := sqlutil.TxStmt(txn, s.selectFilterStmt).QueryRowContext(ctx, localpart, filterID).Scan(&filterData)
 	if err != nil {
 		return err
 	}
@@ -90,7 +91,7 @@ func (s *filterStatements) SelectFilter(
 }
 
 func (s *filterStatements) InsertFilter(
-	ctx context.Context, filter *gomatrixserverlib.Filter, localpart string,
+	ctx context.Context, txn *sql.Tx, filter *gomatrixserverlib.Filter, localpart string,
 ) (filterID string, err error) {
 	var existingFilterID string
 
@@ -111,8 +112,9 @@ func (s *filterStatements) InsertFilter(
 	// This can result in a race condition when two clients try to insert the
 	// same filter and localpart at the same time, however this is not a
 	// problem as both calls will result in the same filterID
-	err = s.selectFilterIDByContentStmt.QueryRowContext(ctx,
-		localpart, filterJSON).Scan(&existingFilterID)
+	err = sqlutil.TxStmt(txn, s.selectFilterIDByContentStmt).QueryRowContext(
+		ctx, localpart, filterJSON,
+	).Scan(&existingFilterID)
 	if err != nil && err != sql.ErrNoRows {
 		return "", err
 	}
@@ -122,7 +124,7 @@ func (s *filterStatements) InsertFilter(
 	}
 
 	// Otherwise insert the filter and return the new ID
-	err = s.insertFilterStmt.QueryRowContext(ctx, filterJSON, localpart).
+	err = sqlutil.TxStmt(txn, s.insertFilterStmt).QueryRowContext(ctx, filterJSON, localpart).
 		Scan(&filterID)
 	return
 }
diff --git a/syncapi/storage/postgres/invites_table.go b/syncapi/storage/postgres/invites_table.go
index 97001ae2..f87ccf96 100644
--- a/syncapi/storage/postgres/invites_table.go
+++ b/syncapi/storage/postgres/invites_table.go
@@ -99,7 +99,7 @@ func (s *inviteEventsStatements) InsertInviteEvent(
 		return
 	}
 
-	err = s.insertInviteEventStmt.QueryRowContext(
+	err = sqlutil.TxStmt(txn, s.insertInviteEventStmt).QueryRowContext(
 		ctx,
 		inviteEvent.RoomID(),
 		inviteEvent.EventID(),
diff --git a/syncapi/storage/postgres/output_room_events_table.go b/syncapi/storage/postgres/output_room_events_table.go
index 20a9ea42..cb092150 100644
--- a/syncapi/storage/postgres/output_room_events_table.go
+++ b/syncapi/storage/postgres/output_room_events_table.go
@@ -222,12 +222,12 @@ func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) {
 	}.Prepare(db)
 }
 
-func (s *outputRoomEventsStatements) UpdateEventJSON(ctx context.Context, event *gomatrixserverlib.HeaderedEvent) error {
+func (s *outputRoomEventsStatements) UpdateEventJSON(ctx context.Context, txn *sql.Tx, event *gomatrixserverlib.HeaderedEvent) error {
 	headeredJSON, err := json.Marshal(event)
 	if err != nil {
 		return err
 	}
-	_, err = s.updateEventJSONStmt.ExecContext(ctx, headeredJSON, event.EventID())
+	_, err = sqlutil.TxStmt(txn, s.updateEventJSONStmt).ExecContext(ctx, headeredJSON, event.EventID())
 	return err
 }
 
diff --git a/syncapi/storage/postgres/output_room_events_topology_table.go b/syncapi/storage/postgres/output_room_events_topology_table.go
index a1fc9b2a..6fab900e 100644
--- a/syncapi/storage/postgres/output_room_events_topology_table.go
+++ b/syncapi/storage/postgres/output_room_events_topology_table.go
@@ -173,7 +173,7 @@ func (s *outputRoomEventsTopologyStatements) SelectEventIDsInRange(
 func (s *outputRoomEventsTopologyStatements) SelectPositionInTopology(
 	ctx context.Context, txn *sql.Tx, eventID string,
 ) (pos, spos types.StreamPosition, err error) {
-	err = s.selectPositionInTopologyStmt.QueryRowContext(ctx, eventID).Scan(&pos, &spos)
+	err = sqlutil.TxStmt(txn, s.selectPositionInTopologyStmt).QueryRowContext(ctx, eventID).Scan(&pos, &spos)
 	return
 }
 
@@ -183,9 +183,9 @@ func (s *outputRoomEventsTopologyStatements) SelectStreamToTopologicalPosition(
 	ctx context.Context, txn *sql.Tx, roomID string, streamPos types.StreamPosition, backwardOrdering bool,
 ) (topoPos types.StreamPosition, err error) {
 	if backwardOrdering {
-		err = s.selectStreamToTopologicalPositionDescStmt.QueryRowContext(ctx, roomID, streamPos).Scan(&topoPos)
+		err = sqlutil.TxStmt(txn, s.selectStreamToTopologicalPositionDescStmt).QueryRowContext(ctx, roomID, streamPos).Scan(&topoPos)
 	} else {
-		err = s.selectStreamToTopologicalPositionAscStmt.QueryRowContext(ctx, roomID, streamPos).Scan(&topoPos)
+		err = sqlutil.TxStmt(txn, s.selectStreamToTopologicalPositionAscStmt).QueryRowContext(ctx, roomID, streamPos).Scan(&topoPos)
 	}
 	return
 }
@@ -193,6 +193,6 @@ func (s *outputRoomEventsTopologyStatements) SelectStreamToTopologicalPosition(
 func (s *outputRoomEventsTopologyStatements) SelectMaxPositionInTopology(
 	ctx context.Context, txn *sql.Tx, roomID string,
 ) (pos types.StreamPosition, spos types.StreamPosition, err error) {
-	err = s.selectMaxPositionInTopologyStmt.QueryRowContext(ctx, roomID).Scan(&pos, &spos)
+	err = sqlutil.TxStmt(txn, s.selectMaxPositionInTopologyStmt).QueryRowContext(ctx, roomID).Scan(&pos, &spos)
 	return
 }
diff --git a/syncapi/storage/postgres/peeks_table.go b/syncapi/storage/postgres/peeks_table.go
index 75eeac98..e20a4882 100644
--- a/syncapi/storage/postgres/peeks_table.go
+++ b/syncapi/storage/postgres/peeks_table.go
@@ -152,9 +152,9 @@ func (s *peekStatements) SelectPeeksInRange(
 }
 
 func (s *peekStatements) SelectPeekingDevices(
-	ctx context.Context,
+	ctx context.Context, txn *sql.Tx,
 ) (peekingDevices map[string][]types.PeekingDevice, err error) {
-	rows, err := s.selectPeekingDevicesStmt.QueryContext(ctx)
+	rows, err := sqlutil.TxStmt(txn, s.selectPeekingDevicesStmt).QueryContext(ctx)
 	if err != nil {
 		return nil, err
 	}
diff --git a/syncapi/storage/postgres/receipt_table.go b/syncapi/storage/postgres/receipt_table.go
index bbddaa93..327a7a37 100644
--- a/syncapi/storage/postgres/receipt_table.go
+++ b/syncapi/storage/postgres/receipt_table.go
@@ -104,9 +104,9 @@ func (r *receiptStatements) UpsertReceipt(ctx context.Context, txn *sql.Tx, room
 	return
 }
 
-func (r *receiptStatements) SelectRoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []types.OutputReceiptEvent, error) {
+func (r *receiptStatements) SelectRoomReceiptsAfter(ctx context.Context, txn *sql.Tx, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []types.OutputReceiptEvent, error) {
 	var lastPos types.StreamPosition
-	rows, err := r.selectRoomReceipts.QueryContext(ctx, pq.Array(roomIDs), streamPos)
+	rows, err := sqlutil.TxStmt(txn, r.selectRoomReceipts).QueryContext(ctx, pq.Array(roomIDs), streamPos)
 	if err != nil {
 		return 0, nil, fmt.Errorf("unable to query room receipts: %w", err)
 	}
diff --git a/syncapi/storage/shared/syncserver.go b/syncapi/storage/shared/syncserver.go
index 47e3a991..a05e6880 100644
--- a/syncapi/storage/shared/syncserver.go
+++ b/syncapi/storage/shared/syncserver.go
@@ -148,7 +148,7 @@ func (d *Database) PeeksInRange(ctx context.Context, userID, deviceID string, r
 }
 
 func (d *Database) RoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []types.OutputReceiptEvent, error) {
-	return d.Receipts.SelectRoomReceiptsAfter(ctx, roomIDs, streamPos)
+	return d.Receipts.SelectRoomReceiptsAfter(ctx, nil, roomIDs, streamPos)
 }
 
 // Events lookups a list of event by their event ID.
@@ -168,15 +168,15 @@ func (d *Database) Events(ctx context.Context, eventIDs []string) ([]*gomatrixse
 }
 
 func (d *Database) AllJoinedUsersInRooms(ctx context.Context) (map[string][]string, error) {
-	return d.CurrentRoomState.SelectJoinedUsers(ctx)
+	return d.CurrentRoomState.SelectJoinedUsers(ctx, nil)
 }
 
 func (d *Database) AllJoinedUsersInRoom(ctx context.Context, roomIDs []string) (map[string][]string, error) {
-	return d.CurrentRoomState.SelectJoinedUsersInRoom(ctx, roomIDs)
+	return d.CurrentRoomState.SelectJoinedUsersInRoom(ctx, nil, roomIDs)
 }
 
 func (d *Database) AllPeekingDevicesInRooms(ctx context.Context) (map[string][]types.PeekingDevice, error) {
-	return d.Peeks.SelectPeekingDevices(ctx)
+	return d.Peeks.SelectPeekingDevices(ctx, nil)
 }
 
 func (d *Database) SharedUsers(ctx context.Context, userID string, otherUserIDs []string) ([]string, error) {
@@ -186,7 +186,7 @@ func (d *Database) SharedUsers(ctx context.Context, userID string, otherUserIDs
 func (d *Database) GetStateEvent(
 	ctx context.Context, roomID, evType, stateKey string,
 ) (*gomatrixserverlib.HeaderedEvent, error) {
-	return d.CurrentRoomState.SelectStateEvent(ctx, roomID, evType, stateKey)
+	return d.CurrentRoomState.SelectStateEvent(ctx, nil, roomID, evType, stateKey)
 }
 
 func (d *Database) GetStateEventsForRoom(
@@ -277,7 +277,7 @@ func (d *Database) GetAccountDataInRange(
 	ctx context.Context, userID string, r types.Range,
 	accountDataFilterPart *gomatrixserverlib.EventFilter,
 ) (map[string][]string, types.StreamPosition, error) {
-	return d.AccountData.SelectAccountDataInRange(ctx, userID, r, accountDataFilterPart)
+	return d.AccountData.SelectAccountDataInRange(ctx, nil, userID, r, accountDataFilterPart)
 }
 
 // UpsertAccountData keeps track of new or updated account data, by saving the type
@@ -484,7 +484,7 @@ func (d *Database) GetEventsInTopologicalRange(
 func (d *Database) BackwardExtremitiesForRoom(
 	ctx context.Context, roomID string,
 ) (backwardExtremities map[string][]string, err error) {
-	return d.BackwardExtremities.SelectBackwardExtremitiesForRoom(ctx, roomID)
+	return d.BackwardExtremities.SelectBackwardExtremitiesForRoom(ctx, nil, roomID)
 }
 
 func (d *Database) MaxTopologicalPosition(
@@ -530,7 +530,7 @@ func (d *Database) StreamToTopologicalPosition(
 func (d *Database) GetFilter(
 	ctx context.Context, target *gomatrixserverlib.Filter, localpart string, filterID string,
 ) error {
-	return d.Filter.SelectFilter(ctx, target, localpart, filterID)
+	return d.Filter.SelectFilter(ctx, nil, target, localpart, filterID)
 }
 
 func (d *Database) PutFilter(
@@ -538,8 +538,8 @@ func (d *Database) PutFilter(
 ) (string, error) {
 	var filterID string
 	var err error
-	err = d.Writer.Do(nil, nil, func(txn *sql.Tx) error {
-		filterID, err = d.Filter.InsertFilter(ctx, filter, localpart)
+	err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
+		filterID, err = d.Filter.InsertFilter(ctx, txn, filter, localpart)
 		return err
 	})
 	return filterID, err
@@ -561,8 +561,8 @@ func (d *Database) RedactEvent(ctx context.Context, redactedEventID string, reda
 	}
 
 	newEvent := eventToRedact.Headered(redactedBecause.RoomVersion)
-	err = d.Writer.Do(nil, nil, func(txn *sql.Tx) error {
-		return d.OutputEvents.UpdateEventJSON(ctx, newEvent)
+	err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
+		return d.OutputEvents.UpdateEventJSON(ctx, txn, newEvent)
 	})
 	return err
 }
@@ -1024,7 +1024,7 @@ func (d *Database) StoreReceipt(ctx context.Context, roomId, receiptType, userId
 }
 
 func (d *Database) GetRoomReceipts(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) ([]types.OutputReceiptEvent, error) {
-	_, receipts, err := d.Receipts.SelectRoomReceiptsAfter(ctx, roomIDs, streamPos)
+	_, receipts, err := d.Receipts.SelectRoomReceiptsAfter(ctx, nil, roomIDs, streamPos)
 	return receipts, err
 }
 
diff --git a/syncapi/storage/sqlite3/account_data_table.go b/syncapi/storage/sqlite3/account_data_table.go
index 21a16dcd..d8967113 100644
--- a/syncapi/storage/sqlite3/account_data_table.go
+++ b/syncapi/storage/sqlite3/account_data_table.go
@@ -91,14 +91,14 @@ func (s *accountDataStatements) InsertAccountData(
 }
 
 func (s *accountDataStatements) SelectAccountDataInRange(
-	ctx context.Context,
+	ctx context.Context, txn *sql.Tx,
 	userID string,
 	r types.Range,
 	filter *gomatrixserverlib.EventFilter,
 ) (data map[string][]string, pos types.StreamPosition, err error) {
 	data = make(map[string][]string)
 	stmt, params, err := prepareWithFilters(
-		s.db, nil, selectAccountDataInRangeSQL,
+		s.db, txn, selectAccountDataInRangeSQL,
 		[]interface{}{
 			userID, r.Low(), r.High(),
 		},
diff --git a/syncapi/storage/sqlite3/backwards_extremities_table.go b/syncapi/storage/sqlite3/backwards_extremities_table.go
index c5674dde..3a5fd6be 100644
--- a/syncapi/storage/sqlite3/backwards_extremities_table.go
+++ b/syncapi/storage/sqlite3/backwards_extremities_table.go
@@ -82,9 +82,9 @@ func (s *backwardExtremitiesStatements) InsertsBackwardExtremity(
 }
 
 func (s *backwardExtremitiesStatements) SelectBackwardExtremitiesForRoom(
-	ctx context.Context, roomID string,
+	ctx context.Context, txn *sql.Tx, roomID string,
 ) (bwExtrems map[string][]string, err error) {
-	rows, err := s.selectBackwardExtremitiesForRoomStmt.QueryContext(ctx, roomID)
+	rows, err := sqlutil.TxStmt(txn, s.selectBackwardExtremitiesForRoomStmt).QueryContext(ctx, roomID)
 	if err != nil {
 		return
 	}
diff --git a/syncapi/storage/sqlite3/current_room_state_table.go b/syncapi/storage/sqlite3/current_room_state_table.go
index bd1271dd..ba6d8126 100644
--- a/syncapi/storage/sqlite3/current_room_state_table.go
+++ b/syncapi/storage/sqlite3/current_room_state_table.go
@@ -163,9 +163,9 @@ func NewSqliteCurrentRoomStateTable(db *sql.DB, streamID *StreamIDStatements) (t
 
 // SelectJoinedUsers returns a map of room ID to a list of joined user IDs.
 func (s *currentRoomStateStatements) SelectJoinedUsers(
-	ctx context.Context,
+	ctx context.Context, txn *sql.Tx,
 ) (map[string][]string, error) {
-	rows, err := s.selectJoinedUsersStmt.QueryContext(ctx)
+	rows, err := sqlutil.TxStmt(txn, s.selectJoinedUsersStmt).QueryContext(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -187,7 +187,7 @@ func (s *currentRoomStateStatements) SelectJoinedUsers(
 
 // SelectJoinedUsersInRoom returns a map of room ID to a list of joined user IDs for a given room.
 func (s *currentRoomStateStatements) SelectJoinedUsersInRoom(
-	ctx context.Context, roomIDs []string,
+	ctx context.Context, txn *sql.Tx, roomIDs []string,
 ) (map[string][]string, error) {
 	query := strings.Replace(selectJoinedUsersInRoomSQL, "($1)", sqlutil.QueryVariadic(len(roomIDs)), 1)
 	params := make([]interface{}, 0, len(roomIDs))
@@ -200,7 +200,7 @@ func (s *currentRoomStateStatements) SelectJoinedUsersInRoom(
 	}
 	defer internal.CloseAndLogIfError(ctx, stmt, "SelectJoinedUsersInRoom: stmt.close() failed")
 
-	rows, err := stmt.QueryContext(ctx, params...)
+	rows, err := sqlutil.TxStmt(txn, stmt).QueryContext(ctx, params...)
 	if err != nil {
 		return nil, err
 	}
@@ -401,9 +401,9 @@ func rowsToEvents(rows *sql.Rows) ([]*gomatrixserverlib.HeaderedEvent, error) {
 }
 
 func (s *currentRoomStateStatements) SelectStateEvent(
-	ctx context.Context, roomID, evType, stateKey string,
+	ctx context.Context, txn *sql.Tx, roomID, evType, stateKey string,
 ) (*gomatrixserverlib.HeaderedEvent, error) {
-	stmt := s.selectStateEventStmt
+	stmt := sqlutil.TxStmt(txn, s.selectStateEventStmt)
 	var res []byte
 	err := stmt.QueryRowContext(ctx, roomID, evType, stateKey).Scan(&res)
 	if err == sql.ErrNoRows {
@@ -429,10 +429,17 @@ func (s *currentRoomStateStatements) SelectSharedUsers(
 		params[k+1] = v
 	}
 
+	var provider sqlutil.QueryProvider
+	if txn == nil {
+		provider = s.db
+	} else {
+		provider = txn
+	}
+
 	result := make([]string, 0, len(otherUserIDs))
 	query := strings.Replace(selectSharedUsersSQL, "($2)", sqlutil.QueryVariadicOffset(len(otherUserIDs), 1), 1)
 	err := sqlutil.RunLimitedVariablesQuery(
-		ctx, query, s.db, params, sqlutil.SQLite3MaxVariables,
+		ctx, query, provider, params, sqlutil.SQLite3MaxVariables,
 		func(rows *sql.Rows) error {
 			var stateKey string
 			for rows.Next() {
diff --git a/syncapi/storage/sqlite3/filter_table.go b/syncapi/storage/sqlite3/filter_table.go
index 6081a48b..5f1e980e 100644
--- a/syncapi/storage/sqlite3/filter_table.go
+++ b/syncapi/storage/sqlite3/filter_table.go
@@ -20,6 +20,7 @@ import (
 	"encoding/json"
 	"fmt"
 
+	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
 	"github.com/matrix-org/gomatrixserverlib"
 )
@@ -77,11 +78,11 @@ func NewSqliteFilterTable(db *sql.DB) (tables.Filter, error) {
 }
 
 func (s *filterStatements) SelectFilter(
-	ctx context.Context, target *gomatrixserverlib.Filter, localpart string, filterID string,
+	ctx context.Context, txn *sql.Tx, target *gomatrixserverlib.Filter, localpart string, filterID string,
 ) error {
 	// Retrieve filter from database (stored as canonical JSON)
 	var filterData []byte
-	err := s.selectFilterStmt.QueryRowContext(ctx, localpart, filterID).Scan(&filterData)
+	err := sqlutil.TxStmt(txn, s.selectFilterStmt).QueryRowContext(ctx, localpart, filterID).Scan(&filterData)
 	if err != nil {
 		return err
 	}
@@ -94,7 +95,7 @@ func (s *filterStatements) SelectFilter(
 }
 
 func (s *filterStatements) InsertFilter(
-	ctx context.Context, filter *gomatrixserverlib.Filter, localpart string,
+	ctx context.Context, txn *sql.Tx, filter *gomatrixserverlib.Filter, localpart string,
 ) (filterID string, err error) {
 	var existingFilterID string
 
@@ -115,8 +116,9 @@ func (s *filterStatements) InsertFilter(
 	// This can result in a race condition when two clients try to insert the
 	// same filter and localpart at the same time, however this is not a
 	// problem as both calls will result in the same filterID
-	err = s.selectFilterIDByContentStmt.QueryRowContext(ctx,
-		localpart, filterJSON).Scan(&existingFilterID)
+	err = sqlutil.TxStmt(txn, s.selectFilterIDByContentStmt).QueryRowContext(
+		ctx, localpart, filterJSON,
+	).Scan(&existingFilterID)
 	if err != nil && err != sql.ErrNoRows {
 		return "", err
 	}
@@ -126,7 +128,7 @@ func (s *filterStatements) InsertFilter(
 	}
 
 	// Otherwise insert the filter and return the new ID
-	res, err := s.insertFilterStmt.ExecContext(ctx, filterJSON, localpart)
+	res, err := sqlutil.TxStmt(txn, s.insertFilterStmt).ExecContext(ctx, filterJSON, localpart)
 	if err != nil {
 		return "", err
 	}
diff --git a/syncapi/storage/sqlite3/notification_data_table.go b/syncapi/storage/sqlite3/notification_data_table.go
index a690ffad..6242898e 100644
--- a/syncapi/storage/sqlite3/notification_data_table.go
+++ b/syncapi/storage/sqlite3/notification_data_table.go
@@ -91,7 +91,12 @@ func (r *notificationDataStatements) SelectUserUnreadCountsForRooms(
 		params[i+1] = roomIDs[i]
 	}
 	sql := strings.Replace(selectUserUnreadNotificationsForRooms, "($2)", sqlutil.QueryVariadicOffset(len(roomIDs), 1), 1)
-	rows, err := r.db.QueryContext(ctx, sql, params...)
+	prep, err := r.db.PrepareContext(ctx, sql)
+	if err != nil {
+		return nil, err
+	}
+	defer internal.CloseAndLogIfError(ctx, prep, "SelectUserUnreadCountsForRooms: prep.close() failed")
+	rows, err := sqlutil.TxStmt(txn, prep).QueryContext(ctx, params...)
 	if err != nil {
 		return nil, err
 	}
diff --git a/syncapi/storage/sqlite3/output_room_events_table.go b/syncapi/storage/sqlite3/output_room_events_table.go
index 6269f4fd..16594302 100644
--- a/syncapi/storage/sqlite3/output_room_events_table.go
+++ b/syncapi/storage/sqlite3/output_room_events_table.go
@@ -164,12 +164,12 @@ func NewSqliteEventsTable(db *sql.DB, streamID *StreamIDStatements) (tables.Even
 	}.Prepare(db)
 }
 
-func (s *outputRoomEventsStatements) UpdateEventJSON(ctx context.Context, event *gomatrixserverlib.HeaderedEvent) error {
+func (s *outputRoomEventsStatements) UpdateEventJSON(ctx context.Context, txn *sql.Tx, event *gomatrixserverlib.HeaderedEvent) error {
 	headeredJSON, err := json.Marshal(event)
 	if err != nil {
 		return err
 	}
-	_, err = s.updateEventJSONStmt.ExecContext(ctx, headeredJSON, event.EventID())
+	_, err = sqlutil.TxStmt(txn, s.updateEventJSONStmt).ExecContext(ctx, headeredJSON, event.EventID())
 	return err
 }
 
@@ -647,7 +647,7 @@ func (s *outputRoomEventsStatements) ReIndex(ctx context.Context, txn *sql.Tx, l
 		return nil, err
 	}
 	defer internal.CloseAndLogIfError(ctx, stmt, "selectEvents: stmt.close() failed")
-	rows, err := stmt.QueryContext(ctx, params...)
+	rows, err := sqlutil.TxStmt(txn, stmt).QueryContext(ctx, params...)
 	if err != nil {
 		return nil, err
 	}
diff --git a/syncapi/storage/sqlite3/output_room_events_topology_table.go b/syncapi/storage/sqlite3/output_room_events_topology_table.go
index b2fb7741..81b26498 100644
--- a/syncapi/storage/sqlite3/output_room_events_topology_table.go
+++ b/syncapi/storage/sqlite3/output_room_events_topology_table.go
@@ -176,9 +176,9 @@ func (s *outputRoomEventsTopologyStatements) SelectStreamToTopologicalPosition(
 	ctx context.Context, txn *sql.Tx, roomID string, streamPos types.StreamPosition, backwardOrdering bool,
 ) (topoPos types.StreamPosition, err error) {
 	if backwardOrdering {
-		err = s.selectStreamToTopologicalPositionDescStmt.QueryRowContext(ctx, roomID, streamPos).Scan(&topoPos)
+		err = sqlutil.TxStmt(txn, s.selectStreamToTopologicalPositionDescStmt).QueryRowContext(ctx, roomID, streamPos).Scan(&topoPos)
 	} else {
-		err = s.selectStreamToTopologicalPositionAscStmt.QueryRowContext(ctx, roomID, streamPos).Scan(&topoPos)
+		err = sqlutil.TxStmt(txn, s.selectStreamToTopologicalPositionAscStmt).QueryRowContext(ctx, roomID, streamPos).Scan(&topoPos)
 	}
 	return
 }
diff --git a/syncapi/storage/sqlite3/peeks_table.go b/syncapi/storage/sqlite3/peeks_table.go
index 5ee86448..4ef51b10 100644
--- a/syncapi/storage/sqlite3/peeks_table.go
+++ b/syncapi/storage/sqlite3/peeks_table.go
@@ -172,9 +172,9 @@ func (s *peekStatements) SelectPeeksInRange(
 }
 
 func (s *peekStatements) SelectPeekingDevices(
-	ctx context.Context,
+	ctx context.Context, txn *sql.Tx,
 ) (peekingDevices map[string][]types.PeekingDevice, err error) {
-	rows, err := s.selectPeekingDevicesStmt.QueryContext(ctx)
+	rows, err := sqlutil.TxStmt(txn, s.selectPeekingDevicesStmt).QueryContext(ctx)
 	if err != nil {
 		return nil, err
 	}
diff --git a/syncapi/storage/sqlite3/receipt_table.go b/syncapi/storage/sqlite3/receipt_table.go
index 31adb005..a4a9b439 100644
--- a/syncapi/storage/sqlite3/receipt_table.go
+++ b/syncapi/storage/sqlite3/receipt_table.go
@@ -108,7 +108,7 @@ func (r *receiptStatements) UpsertReceipt(ctx context.Context, txn *sql.Tx, room
 }
 
 // SelectRoomReceiptsAfter select all receipts for a given room after a specific timestamp
-func (r *receiptStatements) SelectRoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []types.OutputReceiptEvent, error) {
+func (r *receiptStatements) SelectRoomReceiptsAfter(ctx context.Context, txn *sql.Tx, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []types.OutputReceiptEvent, error) {
 	selectSQL := strings.Replace(selectRoomReceipts, "($2)", sqlutil.QueryVariadicOffset(len(roomIDs), 1), 1)
 	var lastPos types.StreamPosition
 	params := make([]interface{}, len(roomIDs)+1)
@@ -116,7 +116,12 @@ func (r *receiptStatements) SelectRoomReceiptsAfter(ctx context.Context, roomIDs
 	for k, v := range roomIDs {
 		params[k+1] = v
 	}
-	rows, err := r.db.QueryContext(ctx, selectSQL, params...)
+	prep, err := r.db.Prepare(selectSQL)
+	if err != nil {
+		return 0, nil, fmt.Errorf("unable to prepare statement: %w", err)
+	}
+	defer internal.CloseAndLogIfError(ctx, prep, "SelectRoomReceiptsAfter: prep.close() failed")
+	rows, err := sqlutil.TxStmt(txn, prep).QueryContext(ctx, params...)
 	if err != nil {
 		return 0, nil, fmt.Errorf("unable to query room receipts: %w", err)
 	}
diff --git a/syncapi/storage/tables/interface.go b/syncapi/storage/tables/interface.go
index 2a6d6fa8..89cb537a 100644
--- a/syncapi/storage/tables/interface.go
+++ b/syncapi/storage/tables/interface.go
@@ -28,7 +28,7 @@ import (
 type AccountData interface {
 	InsertAccountData(ctx context.Context, txn *sql.Tx, userID, roomID, dataType string) (pos types.StreamPosition, err error)
 	// SelectAccountDataInRange returns a map of room ID to a list of `dataType`.
-	SelectAccountDataInRange(ctx context.Context, userID string, r types.Range, accountDataEventFilter *gomatrixserverlib.EventFilter) (data map[string][]string, pos types.StreamPosition, err error)
+	SelectAccountDataInRange(ctx context.Context, txn *sql.Tx, userID string, r types.Range, accountDataEventFilter *gomatrixserverlib.EventFilter) (data map[string][]string, pos types.StreamPosition, err error)
 	SelectMaxAccountDataID(ctx context.Context, txn *sql.Tx) (id int64, err error)
 }
 
@@ -46,7 +46,7 @@ type Peeks interface {
 	DeletePeek(ctx context.Context, txn *sql.Tx, roomID, userID, deviceID string) (streamPos types.StreamPosition, err error)
 	DeletePeeks(ctx context.Context, txn *sql.Tx, roomID, userID string) (streamPos types.StreamPosition, err error)
 	SelectPeeksInRange(ctxt context.Context, txn *sql.Tx, userID, deviceID string, r types.Range) (peeks []types.Peek, err error)
-	SelectPeekingDevices(ctxt context.Context) (peekingDevices map[string][]types.PeekingDevice, err error)
+	SelectPeekingDevices(ctxt context.Context, txn *sql.Tx) (peekingDevices map[string][]types.PeekingDevice, err error)
 	SelectMaxPeekID(ctx context.Context, txn *sql.Tx) (id int64, err error)
 }
 
@@ -68,7 +68,7 @@ type Events interface {
 	// SelectEarlyEvents returns the earliest events in the given room.
 	SelectEarlyEvents(ctx context.Context, txn *sql.Tx, roomID string, r types.Range, eventFilter *gomatrixserverlib.RoomEventFilter) ([]types.StreamEvent, error)
 	SelectEvents(ctx context.Context, txn *sql.Tx, eventIDs []string, filter *gomatrixserverlib.RoomEventFilter, preserveOrder bool) ([]types.StreamEvent, error)
-	UpdateEventJSON(ctx context.Context, event *gomatrixserverlib.HeaderedEvent) error
+	UpdateEventJSON(ctx context.Context, txn *sql.Tx, event *gomatrixserverlib.HeaderedEvent) error
 	// DeleteEventsForRoom removes all event information for a room. This should only be done when removing the room entirely.
 	DeleteEventsForRoom(ctx context.Context, txn *sql.Tx, roomID string) (err error)
 
@@ -98,7 +98,7 @@ type Topology interface {
 }
 
 type CurrentRoomState interface {
-	SelectStateEvent(ctx context.Context, roomID, evType, stateKey string) (*gomatrixserverlib.HeaderedEvent, error)
+	SelectStateEvent(ctx context.Context, txn *sql.Tx, roomID, evType, stateKey string) (*gomatrixserverlib.HeaderedEvent, error)
 	SelectEventsWithEventIDs(ctx context.Context, txn *sql.Tx, eventIDs []string) ([]types.StreamEvent, error)
 	UpsertRoomState(ctx context.Context, txn *sql.Tx, event *gomatrixserverlib.HeaderedEvent, membership *string, addedAt types.StreamPosition) error
 	DeleteRoomStateByEventID(ctx context.Context, txn *sql.Tx, eventID string) error
@@ -110,9 +110,9 @@ type CurrentRoomState interface {
 	// SelectRoomIDsWithAnyMembership returns a map of all memberships for the given user.
 	SelectRoomIDsWithAnyMembership(ctx context.Context, txn *sql.Tx, userID string) (map[string]string, error)
 	// SelectJoinedUsers returns a map of room ID to a list of joined user IDs.
-	SelectJoinedUsers(ctx context.Context) (map[string][]string, error)
+	SelectJoinedUsers(ctx context.Context, txn *sql.Tx) (map[string][]string, error)
 	// SelectJoinedUsersInRoom returns a map of room ID to a list of joined user IDs for a given room.
-	SelectJoinedUsersInRoom(ctx context.Context, roomIDs []string) (map[string][]string, error)
+	SelectJoinedUsersInRoom(ctx context.Context, txn *sql.Tx, roomIDs []string) (map[string][]string, error)
 	// SelectSharedUsers returns a subset of otherUserIDs that share a room with userID.
 	SelectSharedUsers(ctx context.Context, txn *sql.Tx, userID string, otherUserIDs []string) ([]string, error)
 }
@@ -142,7 +142,7 @@ type BackwardsExtremities interface {
 	// InsertsBackwardExtremity inserts a new backwards extremity.
 	InsertsBackwardExtremity(ctx context.Context, txn *sql.Tx, roomID, eventID string, prevEventID string) (err error)
 	// SelectBackwardExtremitiesForRoom retrieves all backwards extremities for the room, as a map of event_id to list of prev_event_ids.
-	SelectBackwardExtremitiesForRoom(ctx context.Context, roomID string) (bwExtrems map[string][]string, err error)
+	SelectBackwardExtremitiesForRoom(ctx context.Context, txn *sql.Tx, roomID string) (bwExtrems map[string][]string, err error)
 	// DeleteBackwardExtremity removes a backwards extremity for a room, if one existed.
 	DeleteBackwardExtremity(ctx context.Context, txn *sql.Tx, roomID, knownEventID string) (err error)
 }
@@ -172,13 +172,13 @@ type SendToDevice interface {
 }
 
 type Filter interface {
-	SelectFilter(ctx context.Context, target *gomatrixserverlib.Filter, localpart string, filterID string) error
-	InsertFilter(ctx context.Context, filter *gomatrixserverlib.Filter, localpart string) (filterID string, err error)
+	SelectFilter(ctx context.Context, txn *sql.Tx, target *gomatrixserverlib.Filter, localpart string, filterID string) error
+	InsertFilter(ctx context.Context, txn *sql.Tx, filter *gomatrixserverlib.Filter, localpart string) (filterID string, err error)
 }
 
 type Receipts interface {
 	UpsertReceipt(ctx context.Context, txn *sql.Tx, roomId, receiptType, userId, eventId string, timestamp gomatrixserverlib.Timestamp) (pos types.StreamPosition, err error)
-	SelectRoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []types.OutputReceiptEvent, error)
+	SelectRoomReceiptsAfter(ctx context.Context, txn *sql.Tx, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []types.OutputReceiptEvent, error)
 	SelectMaxReceiptID(ctx context.Context, txn *sql.Tx) (id int64, err error)
 }