Ensure only one transaction is used for RS input per room (#2178)

* Ensure the input API only uses a single transaction

* Remove more of the dead query API call

* Tidy up

* Fix tests hopefully

* Don't do unnecessary work for rooms that don't exist

* Improve error, fix another case where transaction wasn't used properly

* Add a unit test for checking single transaction on RS input API

* Fix logic oops when deciding whether to use a transaction in storeEvent

3 files changed, 30 insertions, 4 deletions

diff --git a/roomserver/storage/postgres/event_json_table.go b/roomserver/storage/postgres/event_json_table.go
index 433e445d..b3220eff 100644
--- a/roomserver/storage/postgres/event_json_table.go
+++ b/roomserver/storage/postgres/event_json_table.go
@@ -76,7 +76,8 @@ func prepareEventJSONTable(db *sql.DB) (tables.EventJSON, error) {
 func (s *eventJSONStatements) InsertEventJSON(
 	ctx context.Context, txn *sql.Tx, eventNID types.EventNID, eventJSON []byte,
 ) error {
-	_, err := s.insertEventJSONStmt.ExecContext(ctx, int64(eventNID), eventJSON)
+	stmt := sqlutil.TxStmt(txn, s.insertEventJSONStmt)
+	_, err := stmt.ExecContext(ctx, int64(eventNID), eventJSON)
 	return err
 }
 
diff --git a/roomserver/storage/shared/room_updater.go b/roomserver/storage/shared/room_updater.go
index fc75a260..89b878b9 100644
--- a/roomserver/storage/shared/room_updater.go
+++ b/roomserver/storage/shared/room_updater.go
@@ -16,6 +16,7 @@ type RoomUpdater struct {
 	latestEvents            []types.StateAtEventAndReference
 	lastEventIDSent         string
 	currentStateSnapshotNID types.StateSnapshotNID
+	roomExists              bool
 }
 
 func rollback(txn *sql.Tx) {
@@ -33,7 +34,7 @@ func NewRoomUpdater(ctx context.Context, d *Database, txn *sql.Tx, roomInfo *typ
 	// succeed, processing a create event which creates the room, or it won't.
 	if roomInfo == nil {
 		return &RoomUpdater{
-			transaction{ctx, txn}, d, nil, nil, "", 0,
+			transaction{ctx, txn}, d, nil, nil, "", 0, false,
 		}, nil
 	}
 
@@ -57,10 +58,15 @@ func NewRoomUpdater(ctx context.Context, d *Database, txn *sql.Tx, roomInfo *typ
 		}
 	}
 	return &RoomUpdater{
-		transaction{ctx, txn}, d, roomInfo, stateAndRefs, lastEventIDSent, currentStateSnapshotNID,
+		transaction{ctx, txn}, d, roomInfo, stateAndRefs, lastEventIDSent, currentStateSnapshotNID, true,
 	}, nil
 }
 
+// RoomExists returns true if the room exists and false otherwise.
+func (u *RoomUpdater) RoomExists() bool {
+	return u.roomExists
+}
+
 // Implements sqlutil.Transaction
 func (u *RoomUpdater) Commit() error {
 	if u.txn == nil { // SQLite mode probably
@@ -97,6 +103,25 @@ func (u *RoomUpdater) CurrentStateSnapshotNID() types.StateSnapshotNID {
 	return u.currentStateSnapshotNID
 }
 
+func (u *RoomUpdater) MissingAuthPrevEvents(
+	ctx context.Context, e *gomatrixserverlib.Event,
+) (missingAuth, missingPrev []string, err error) {
+	for _, authEventID := range e.AuthEventIDs() {
+		if nids, err := u.EventNIDs(ctx, []string{authEventID}); err != nil || len(nids) == 0 {
+			missingAuth = append(missingAuth, authEventID)
+		}
+	}
+
+	for _, prevEventID := range e.PrevEventIDs() {
+		state, err := u.StateAtEventIDs(ctx, []string{prevEventID})
+		if err != nil || len(state) == 0 || (!state[0].IsCreate() && state[0].BeforeStateSnapshotNID == 0) {
+			missingPrev = append(missingPrev, prevEventID)
+		}
+	}
+
+	return
+}
+
 // StorePreviousEvents implements types.RoomRecentEventsUpdater - This must be called from a Writer
 func (u *RoomUpdater) StorePreviousEvents(eventNID types.EventNID, previousEventReferences []gomatrixserverlib.EventReference) error {
 	return u.d.Writer.Do(u.d.DB, u.txn, func(txn *sql.Tx) error {
diff --git a/roomserver/storage/shared/storage.go b/roomserver/storage/shared/storage.go
index 8319de26..e96c77af 100644
--- a/roomserver/storage/shared/storage.go
+++ b/roomserver/storage/shared/storage.go
@@ -553,7 +553,7 @@ func (d *Database) storeEvent(
 		err              error
 	)
 	var txn *sql.Tx
-	if updater != nil {
+	if updater != nil && updater.txn != nil {
 		txn = updater.txn
 	}
 	err = d.Writer.Do(d.DB, txn, func(txn *sql.Tx) error {